Native cql changes

.github/workflows/continuous-integration.yaml

@@ -6,7 +6,7 @@ name: Continuous Integration
 # * manual trigger
 on:
   push:
-    branches: [ "main" ]
+    branches: [ "main"]
 
   pull_request:
     branches: [ "main" ]

pom.xml

@@ -77,6 +77,26 @@
       <groupId>io.quarkus</groupId>
       <artifactId>quarkus-rest-client-reactive-jackson</artifactId>
     </dependency>
+    <dependency>
+      <groupId>com.datastax.oss</groupId>
+      <artifactId>java-driver-core</artifactId>
+      <version>${driver.version}</version>
+      <exclusions>
+        <exclusion>
+          <groupId>io.dropwizard.metrics</groupId>
+          <artifactId>metrics-core</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>org.hdrhistogram</groupId>
+          <artifactId>HdrHistogram</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <dependency>
+      <groupId>com.datastax.oss</groupId>
+      <artifactId>java-driver-metrics-micrometer</artifactId>
+      <version>${driver.version}</version>
+    </dependency>
     <dependency>
       <groupId>io.quarkus</groupId>
       <artifactId>quarkus-junit5</artifactId>
@@ -103,11 +123,6 @@
       <artifactId>assertj-core</artifactId>
       <scope>test</scope>
     </dependency>
-    <dependency>
-      <groupId>com.datastax.oss</groupId>
-      <artifactId>java-driver-core</artifactId>
-      <scope>test</scope>
-    </dependency>
     <dependency>
       <groupId>io.stargate</groupId>
       <artifactId>sgv2-quarkus-common</artifactId>

src/main/java/com/github/benmanes/caffeine/cache/ReflectionRegistration.java

@@ -0,0 +1,8 @@
+package com.github.benmanes.caffeine.cache;
+
+import io.quarkus.runtime.annotations.RegisterForReflection;
+
+@RegisterForReflection(targets = {com.github.benmanes.caffeine.cache.PSAMS.class})
+public class ReflectionRegistration {
+  // This class is used only for annotation processing during build
+}

src/main/java/io/stargate/sgv2/jsonapi/api/model/command/CommandContext.java

@@ -4,7 +4,7 @@
 import com.fasterxml.jackson.databind.node.JsonNodeFactory;
 import io.stargate.sgv2.jsonapi.api.model.command.clause.sort.SortClause;
 import io.stargate.sgv2.jsonapi.api.model.command.clause.update.UpdateClause;
-import io.stargate.sgv2.jsonapi.service.bridge.executor.CollectionSettings;
+import io.stargate.sgv2.jsonapi.service.cqldriver.executor.CollectionSettings;
 import io.stargate.sgv2.jsonapi.service.embedding.DataVectorizer;
 import io.stargate.sgv2.jsonapi.service.embedding.operation.EmbeddingService;
 import java.util.List;

src/main/java/io/stargate/sgv2/jsonapi/api/v1/CollectionResource.java

@@ -20,7 +20,7 @@
 import io.stargate.sgv2.jsonapi.config.constants.OpenApiConstants;
 import io.stargate.sgv2.jsonapi.exception.JsonApiException;
 import io.stargate.sgv2.jsonapi.exception.mappers.ThrowableCommandResultSupplier;
-import io.stargate.sgv2.jsonapi.service.bridge.executor.SchemaCache;
+import io.stargate.sgv2.jsonapi.service.cqldriver.executor.SchemaCache;
 import io.stargate.sgv2.jsonapi.service.embedding.operation.EmbeddingService;
 import io.stargate.sgv2.jsonapi.service.embedding.operation.EmbeddingServiceCache;
 import io.stargate.sgv2.jsonapi.service.processor.MeteredCommandProcessor;

src/main/java/io/stargate/sgv2/jsonapi/config/ConsistencyLevelConverter.java

@@ -0,0 +1,20 @@
+package io.stargate.sgv2.jsonapi.config;
+
+import com.datastax.oss.driver.api.core.ConsistencyLevel;
+import com.datastax.oss.driver.api.core.DefaultConsistencyLevel;
+import org.eclipse.microprofile.config.spi.Converter;
+
+/**
+ * Converts a string to a {@link ConsistencyLevel}, used in {@link OperationsConfig.QueriesConfig}.
+ */
+public class ConsistencyLevelConverter implements Converter<ConsistencyLevel> {
+  /**
+   * @param value the string representation of a property value
+   * @return the converted ConsistencyLevel
+   */
+  @Override
+  public ConsistencyLevel convert(String value)
+      throws IllegalArgumentException, NullPointerException {
+    return DefaultConsistencyLevel.valueOf(value.toUpperCase());
+  }
+}

src/main/java/io/stargate/sgv2/jsonapi/config/OperationsConfig.java

@@ -17,12 +17,18 @@
 
 package io.stargate.sgv2.jsonapi.config;
 
+import static io.stargate.sgv2.jsonapi.service.cqldriver.CQLSessionCache.CASSANDRA;
+
+import com.datastax.oss.driver.api.core.ConsistencyLevel;
 import io.smallrye.config.ConfigMapping;
+import io.smallrye.config.WithConverter;
 import io.smallrye.config.WithDefault;
 import jakarta.validation.Valid;
 import jakarta.validation.constraints.Max;
 import jakarta.validation.constraints.NotNull;
 import jakarta.validation.constraints.Positive;
+import java.util.List;
+import javax.annotation.Nullable;
 
 /** Configuration for the operation execution. */
 @ConfigMapping(prefix = "stargate.jsonapi.operations")
@@ -109,4 +115,93 @@ interface LwtConfig {
     @WithDefault("3")
     int retries();
   }
+
+  /** Cassandra/AstraDB related configurations. */
+  @NotNull
+  @Valid
+  DatabaseConfig databaseConfig();
+
+  interface DatabaseConfig {
+
+    /** Database type can be <code>cassandra</code> or <code>astra</code>. */
+    @WithDefault(CASSANDRA)
+    String type();
+
+    /** Username when connecting to cassandra database (when type is <code>cassandra</code>) */
+    @Nullable
+    @WithDefault("cassandra")
+    String userName();
+
+    /** Password when connecting to cassandra database (when type is <code>cassandra</code>) */
+    @Nullable
+    @WithDefault("cassandra")
+    String password();
+
+    /** Fixed Token used for Integration Test authentication */
+    @Nullable
+    @WithDefault("not in tests")
+    String fixedToken();
+
+    /** Cassandra contact points (when type is <code>cassandra</code>) */
+    @Nullable
+    @WithDefault("127.0.0.1")
+    List<String> cassandraEndPoints();
+
+    /** Cassandra contact points (when type is <code>cassandra</code>) */
+    @Nullable
+    @WithDefault("9042")
+    int cassandraPort();
+
+    /** Local datacenter that the driver must be configured with */
+    @NotNull
+    @WithDefault("datacenter1")
+    String localDatacenter();
+
+    /** Time to live for CQLSession in cache in seconds. */
+    @WithDefault("300")
+    long sessionCacheTtlSeconds();
+
+    /** Maximum number of CQLSessions in cache. */
+    @WithDefault("100")
+    long sessionCacheMaxSize();
+  }
+
+  /** Query consistency related configs. */
+  @NotNull
+  @Valid
+  QueriesConfig queriesConfig();
+
+  interface QueriesConfig {
+
+    /** @return Settings for the consistency level. */
+    @Valid
+    ConsistencyConfig consistency();
+
+    /** @return Serial Consistency for queries. */
+    @WithDefault("SERIAL")
+    @WithConverter(ConsistencyLevelConverter.class)
+    ConsistencyLevel serialConsistency();
+
+    /** @return Settings for the consistency level. */
+    interface ConsistencyConfig {
+
+      /** @return Consistency for queries making schema changes. */
+      @WithDefault("LOCAL_QUORUM")
+      @NotNull
+      @WithConverter(ConsistencyLevelConverter.class)
+      ConsistencyLevel schemaChanges();
+
+      /** @return Consistency for queries writing the data. */
+      @WithDefault("LOCAL_QUORUM")
+      @NotNull
+      @WithConverter(ConsistencyLevelConverter.class)
+      ConsistencyLevel writes();
+
+      /** @return Consistency for queries reading the data. */
+      @WithDefault("LOCAL_QUORUM")
+      @NotNull
+      @WithConverter(ConsistencyLevelConverter.class)
+      ConsistencyLevel reads();
+    }
+  }
 }

src/main/java/io/stargate/sgv2/jsonapi/exception/JsonApiException.java

@@ -63,17 +63,15 @@ public CommandResult get() {
   public CommandResult.Error getCommandResultError(String message) {
     Map<String, Object> fieldsForMetricsTag =
         Map.of("errorCode", errorCode.name(), "exceptionClass", this.getClass().getSimpleName());
-    Map<String, Object> fields = null;
     SmallRyeConfig config = ConfigProvider.getConfig().unwrap(SmallRyeConfig.class);
     // enable debug mode for unit tests, since it can not be injected
     DebugModeConfig debugModeConfig = config.getConfigMapping(DebugModeConfig.class);
     final boolean debugEnabled = debugModeConfig.enabled();
-    if (debugEnabled) {
-      fields =
-          Map.of("errorCode", errorCode.name(), "exceptionClass", this.getClass().getSimpleName());
-    } else {
-      fields = Map.of("errorCode", errorCode.name());
-    }
+    final Map<String, Object> fields =
+        debugEnabled
+            ? Map.of(
+                "errorCode", errorCode.name(), "exceptionClass", this.getClass().getSimpleName())
+            : Map.of("errorCode", errorCode.name());
     return new CommandResult.Error(message, fieldsForMetricsTag, fields, Response.Status.OK);
   }
 

src/main/java/io/stargate/sgv2/jsonapi/exception/mappers/ThrowableToErrorMapper.java

@@ -1,7 +1,15 @@
 package io.stargate.sgv2.jsonapi.exception.mappers;
 
+import com.datastax.oss.driver.api.core.AllNodesFailedException;
+import com.datastax.oss.driver.api.core.DriverException;
+import com.datastax.oss.driver.api.core.DriverTimeoutException;
+import com.datastax.oss.driver.api.core.NoNodeAvailableException;
+import com.datastax.oss.driver.api.core.NodeUnavailableException;
+import com.datastax.oss.driver.api.core.servererrors.QueryValidationException;
+import com.datastax.oss.driver.api.core.servererrors.WriteTimeoutException;
 import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
+import io.quarkus.security.UnauthorizedException;
 import io.smallrye.config.SmallRyeConfig;
 import io.stargate.sgv2.jsonapi.api.model.command.CommandResult;
 import io.stargate.sgv2.jsonapi.config.DebugModeConfig;
@@ -17,7 +25,6 @@
  * implementation.
  */
 public final class ThrowableToErrorMapper {
-
   private static final BiFunction<Throwable, String, CommandResult.Error> MAPPER_WITH_MESSAGE =
       (throwable, message) -> {
         // if our own exception, shortcut
@@ -28,14 +35,17 @@ public final class ThrowableToErrorMapper {
         SmallRyeConfig config = ConfigProvider.getConfig().unwrap(SmallRyeConfig.class);
         DebugModeConfig debugModeConfig = config.getConfigMapping(DebugModeConfig.class);
         final boolean debugEnabled = debugModeConfig.enabled();
+        final Map<String, Object> fields =
+            debugEnabled ? Map.of("exceptionClass", throwable.getClass().getSimpleName()) : null;
+        final Map<String, Object> fieldsForMetricsTag =
+            Map.of("exceptionClass", throwable.getClass().getSimpleName());
         if (throwable instanceof StatusRuntimeException sre) {
-          Map<String, Object> fields =
-              debugEnabled ? Map.of("exceptionClass", throwable.getClass().getSimpleName()) : null;
-          Map<String, Object> fieldsForMetricsTag =
-              Map.of("exceptionClass", throwable.getClass().getSimpleName());
           if (sre.getStatus().getCode() == Status.Code.UNAUTHENTICATED) {
             return new CommandResult.Error(
-                message, fieldsForMetricsTag, fields, Response.Status.UNAUTHORIZED);
+                "UNAUTHENTICATED: Invalid token",
+                fieldsForMetricsTag,
+                fields,
+                Response.Status.UNAUTHORIZED);
           } else if (sre.getStatus().getCode() == Status.Code.INTERNAL) {
             return new CommandResult.Error(
                 message, fieldsForMetricsTag, fields, Response.Status.INTERNAL_SERVER_ERROR);
@@ -47,11 +57,30 @@ public final class ThrowableToErrorMapper {
                 message, fieldsForMetricsTag, fields, Response.Status.GATEWAY_TIMEOUT);
           }
         }
-        // add error code as error field
-        Map<String, Object> fields =
-            debugEnabled ? Map.of("exceptionClass", throwable.getClass().getSimpleName()) : null;
-        Map<String, Object> fieldsForMetricsTag =
-            Map.of("exceptionClass", throwable.getClass().getSimpleName());
+        if (throwable instanceof UnauthorizedException
+            || throwable
+                instanceof com.datastax.oss.driver.api.core.servererrors.UnauthorizedException) {
+          return new CommandResult.Error(
+              "UNAUTHENTICATED: Invalid token",
+              fieldsForMetricsTag,
+              fields,
+              Response.Status.UNAUTHORIZED);
+        } else if (throwable instanceof QueryValidationException) {
+          if (message.contains("vector<float,")) { // TODO is there a better way?
+            message = "Mismatched vector dimension";
+          }
+          return new CommandResult.Error(message, fieldsForMetricsTag, fields, Response.Status.OK);
+        } else if (throwable instanceof NodeUnavailableException
+            || throwable instanceof DriverException
+            || throwable instanceof AllNodesFailedException
+            || throwable instanceof NoNodeAvailableException) {
+          return new CommandResult.Error(
+              message, fieldsForMetricsTag, fields, Response.Status.INTERNAL_SERVER_ERROR);
+        } else if (throwable instanceof DriverTimeoutException
+            || throwable instanceof WriteTimeoutException) {
+          return new CommandResult.Error(
+              message, fieldsForMetricsTag, fields, Response.Status.GATEWAY_TIMEOUT);
+        }
         return new CommandResult.Error(message, fieldsForMetricsTag, fields, Response.Status.OK);
       };
 
@@ -61,7 +90,6 @@ public final class ThrowableToErrorMapper {
         if (message == null) {
           message = "Unexpected exception occurred.";
         }
-
         return MAPPER_WITH_MESSAGE.apply(throwable, message);
       };