Added Insert One and Create Collection commands

pom.xml

@@ -75,6 +75,22 @@
       <artifactId>assertj-core</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.immutables</groupId>
+      <artifactId>value</artifactId>
+      <scope>provided</scope>
+      <version>2.9.0</version>
+    </dependency>
+    <dependency>
+      <groupId>com.datastax.oss</groupId>
+      <artifactId>java-driver-core</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.javatuples</groupId>
+      <artifactId>javatuples</artifactId>
+      <version>1.2</version>
+    </dependency>
     <dependency>
       <groupId>io.stargate</groupId>
       <artifactId>sgv2-quarkus-common</artifactId>
@@ -133,12 +149,19 @@
   </build>
   <profiles>
     <profile>
-      <id>cassandra-40</id>
+      <id>dse-68</id>
       <activation>
         <activeByDefault>true</activeByDefault>
       </activation>
       <properties>
+        <stargate.int-test.cassandra.image>datastax/dse-server</stargate.int-test.cassandra.image>
+        <stargate.int-test.cassandra.image-tag>6.8.28</stargate.int-test.cassandra.image-tag>
+        <stargate.int-test.coordinator.image>stargateio/coordinator-dse-68</stargate.int-test.coordinator.image>
         <stargate.int-test.coordinator.image-tag>v${stargate.version}</stargate.int-test.coordinator.image-tag>
+        <stargate.int-test.cluster.name>dse-${stargate.int-test.cassandra.image-tag}-cluster</stargate.int-test.cluster.name>
+        <stargate.int-test.cluster.version>6.8</stargate.int-test.cluster.version>
+        <stargate.int-test.cluster.dse>true</stargate.int-test.cluster.dse>
+        <stargate.int-test.cassandra.auth-enabled>true</stargate.int-test.cassandra.auth-enabled>
       </properties>
     </profile>
     <profile>
@@ -153,7 +176,14 @@
         <quarkus.package.type>native</quarkus.package.type>
         <quarkus.container-image.name>docsapi-native</quarkus.container-image.name>
         <quarkus.native.native-image-xmx>6G</quarkus.native.native-image-xmx>
+        <stargate.int-test.cassandra.image>datastax/dse-server</stargate.int-test.cassandra.image>
+        <stargate.int-test.cassandra.image-tag>6.8.28</stargate.int-test.cassandra.image-tag>
+        <stargate.int-test.coordinator.image>stargateio/coordinator-dse-68</stargate.int-test.coordinator.image>
         <stargate.int-test.coordinator.image-tag>v${stargate.version}</stargate.int-test.coordinator.image-tag>
+        <stargate.int-test.cluster.name>dse-${stargate.int-test.cassandra.image-tag}-cluster</stargate.int-test.cluster.name>
+        <stargate.int-test.cluster.version>6.8</stargate.int-test.cluster.version>
+        <stargate.int-test.cluster.dse>true</stargate.int-test.cluster.dse>
+        <stargate.int-test.cassandra.auth-enabled>true</stargate.int-test.cassandra.auth-enabled>
       </properties>
     </profile>
   </profiles>

src/main/java/io/stargate/sgv3/docsapi/StargateDocsV3Api.java

@@ -134,6 +134,17 @@
                         ]
                       }
                       """),
+              @ExampleObject(
+                  name = "resultCreateCollection",
+                  summary = "Create result",
+                  value =
+                      """
+                      {
+                        "status": {
+                            "ok": 1
+                        }
+                      }
+                      """),
             }))
 public class StargateDocsV3Api extends Application {
 

src/main/java/io/stargate/sgv3/docsapi/api/model/command/Command.java

@@ -2,6 +2,7 @@
 
 import com.fasterxml.jackson.annotation.JsonSubTypes;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import io.stargate.sgv3.docsapi.api.model.command.impl.CreateCollectionCommand;
 import io.stargate.sgv3.docsapi.api.model.command.impl.FindOneCommand;
 import io.stargate.sgv3.docsapi.api.model.command.impl.InsertOneCommand;
 
@@ -27,6 +28,7 @@
     include = JsonTypeInfo.As.WRAPPER_OBJECT,
     property = "commandName")
 @JsonSubTypes({
+  @JsonSubTypes.Type(value = CreateCollectionCommand.class),
   @JsonSubTypes.Type(value = FindOneCommand.class),
   @JsonSubTypes.Type(value = InsertOneCommand.class),
 })

src/main/java/io/stargate/sgv3/docsapi/api/model/command/CommandStatus.java

@@ -4,5 +4,7 @@
 
 public enum CommandStatus {
   @JsonProperty("insertedIds")
-  INSERTED_IDS;
+  INSERTED_IDS,
+  @JsonProperty("ok")
+  CREATE_COLLECTION;
 }

src/main/java/io/stargate/sgv3/docsapi/api/model/command/SchemaChangeCommand.java

@@ -0,0 +1,3 @@
+package io.stargate.sgv3.docsapi.api.model.command;
+
+public interface SchemaChangeCommand extends Command {}

src/main/java/io/stargate/sgv3/docsapi/api/model/command/impl/CreateCollectionCommand.java

@@ -0,0 +1,18 @@
+package io.stargate.sgv3.docsapi.api.model.command.impl;
+
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import io.stargate.sgv3.docsapi.api.model.command.SchemaChangeCommand;
+import javax.validation.constraints.NotNull;
+import org.eclipse.microprofile.openapi.annotations.enums.SchemaType;
+import org.eclipse.microprofile.openapi.annotations.media.Schema;
+
+@Schema(description = "Command that creates a collection.")
+@JsonTypeName("createCollection")
+public record CreateCollectionCommand(
+    @NotNull
+        @Schema(
+            description = "Name of the collection",
+            implementation = Object.class,
+            type = SchemaType.OBJECT)
+        String name)
+    implements SchemaChangeCommand {}

src/main/java/io/stargate/sgv3/docsapi/api/v3/DatabaseResource.java

@@ -1,12 +1,30 @@
 package io.stargate.sgv3.docsapi.api.v3;
 
 import io.smallrye.mutiny.Uni;
+import io.stargate.sgv3.docsapi.api.model.command.Command;
+import io.stargate.sgv3.docsapi.api.model.command.CommandContext;
+import io.stargate.sgv3.docsapi.api.model.command.CommandResult;
+import io.stargate.sgv3.docsapi.api.model.command.impl.CreateCollectionCommand;
 import io.stargate.sgv3.docsapi.config.constants.OpenApiConstants;
+import io.stargate.sgv3.docsapi.service.processor.CommandProcessor;
+import javax.inject.Inject;
+import javax.validation.Valid;
+import javax.validation.constraints.NotNull;
 import javax.ws.rs.Consumes;
 import javax.ws.rs.POST;
 import javax.ws.rs.Path;
+import javax.ws.rs.PathParam;
 import javax.ws.rs.Produces;
 import javax.ws.rs.core.MediaType;
+import org.eclipse.microprofile.openapi.annotations.Operation;
+import org.eclipse.microprofile.openapi.annotations.media.Content;
+import org.eclipse.microprofile.openapi.annotations.media.ExampleObject;
+import org.eclipse.microprofile.openapi.annotations.media.Schema;
+import org.eclipse.microprofile.openapi.annotations.parameters.Parameter;
+import org.eclipse.microprofile.openapi.annotations.parameters.Parameters;
+import org.eclipse.microprofile.openapi.annotations.parameters.RequestBody;
+import org.eclipse.microprofile.openapi.annotations.responses.APIResponse;
+import org.eclipse.microprofile.openapi.annotations.responses.APIResponses;
 import org.eclipse.microprofile.openapi.annotations.security.SecurityRequirement;
 import org.eclipse.microprofile.openapi.annotations.tags.Tag;
 import org.jboss.resteasy.reactive.RestResponse;
@@ -20,8 +38,49 @@ public class DatabaseResource {
 
   public static final String BASE_PATH = "/v3/{database}";
 
+  private final CommandProcessor commandProcessor;
+
+  @Inject
+  public DatabaseResource(CommandProcessor commandProcessor) {
+    this.commandProcessor = commandProcessor;
+  }
+
+  @Operation(
+      summary = "Execute command",
+      description = "Executes a single command against a collection.")
+  @Parameters(value = {@Parameter(name = "database", ref = "database")})
+  @RequestBody(
+      content =
+          @Content(
+              mediaType = MediaType.APPLICATION_JSON,
+              schema = @Schema(anyOf = {CreateCollectionCommand.class}),
+              examples = {
+                @ExampleObject(ref = "resultCreateCollection"),
+              }))
+  @APIResponses(
+      @APIResponse(
+          responseCode = "200",
+          description =
+              "Call successful. Returns result of the command execution. Note that in case of errors, response code remains `HTTP 200`.",
+          content =
+              @Content(
+                  mediaType = MediaType.APPLICATION_JSON,
+                  schema = @Schema(implementation = CommandResult.class),
+                  examples = {
+                    @ExampleObject(ref = "resultCreateCollection"),
+                    @ExampleObject(ref = "resultError"),
+                  })))
   @POST
-  public Uni<RestResponse<?>> postCommand() {
-    return Uni.createFrom().item(RestResponse.ok());
+  public Uni<RestResponse<CommandResult>> postCommand(
+      @NotNull @Valid Command command, @PathParam("database") String database) {
+
+    // create context
+    CommandContext commandContext = new CommandContext(database, null);
+
+    // call processor
+    return commandProcessor
+        .processCommand(commandContext, command)
+        // map to 2xx always
+        .map(RestResponse::ok);
   }
 }

src/main/java/io/stargate/sgv3/docsapi/exception/ErrorCode.java

@@ -14,7 +14,11 @@ public enum ErrorCode {
 
   SHRED_NO_MD5("MD5 Hash algorithm not available"),
 
-  SHRED_UNRECOGNIZED_NODE_TYPE("Unrecognized JSON node type in input document");
+  SHRED_UNRECOGNIZED_NODE_TYPE("Unrecognized JSON node type in input document"),
+
+  CREATE_COLLECTION_FAILED("Create collection failed."),
+
+  INSERT_ERROR("Unable to insert the document");
 
   private final String message;
 

src/main/java/io/stargate/sgv3/docsapi/service/bridge/executor/QueryExecutor.java

@@ -0,0 +1,74 @@
+package io.stargate.sgv3.docsapi.service.bridge.executor;
+
+import com.google.protobuf.ByteString;
+import com.google.protobuf.BytesValue;
+import com.google.protobuf.Int32Value;
+import io.smallrye.mutiny.Uni;
+import io.stargate.bridge.proto.QueryOuterClass;
+import io.stargate.sgv2.api.common.StargateRequestInfo;
+import io.stargate.sgv2.api.common.config.QueriesConfig;
+import java.util.Base64;
+import javax.enterprise.context.ApplicationScoped;
+import javax.inject.Inject;
+
+@ApplicationScoped
+public class QueryExecutor {
+
+  private final QueriesConfig queriesConfig;
+
+  private final StargateRequestInfo requestInfo;
+
+  @Inject
+  public QueryExecutor(QueriesConfig queriesConfig, StargateRequestInfo requestInfo) {
+    this.queriesConfig = queriesConfig;
+    this.requestInfo = requestInfo;
+  }
+
+  public Uni<QueryOuterClass.ResultSet> writeDocument(QueryOuterClass.Query query) {
+    return queryBridge(query, null);
+  }
+
+  public Uni<QueryOuterClass.ResultSet> execute(QueryOuterClass.Query query) {
+    return queryBridge(query, null);
+  }
+
+  private Uni<QueryOuterClass.ResultSet> queryBridge(
+      QueryOuterClass.Query query, String pagingState) {
+    // construct initial state for the query
+    BytesValue pagingStateValue =
+        pagingState != null ? BytesValue.of(ByteString.copyFrom(decodeBase64(pagingState))) : null;
+
+    QueryState state = ImmutableQueryState.of(100, pagingStateValue);
+    QueryOuterClass.Consistency consistency = queriesConfig.consistency().reads();
+    QueryOuterClass.ConsistencyValue.Builder consistencyValue =
+        QueryOuterClass.ConsistencyValue.newBuilder().setValue(consistency);
+    QueryOuterClass.QueryParameters.Builder params =
+        QueryOuterClass.QueryParameters.newBuilder()
+            .setConsistency(consistencyValue)
+            .setPageSize(Int32Value.of(state.pageSize()));
+
+    // if we have paging state, set
+    if (null != state.pagingState()) {
+      params.setPagingState(state.pagingState());
+    }
+
+    // final query is same as the original, just with different params
+    QueryOuterClass.Query finalQuery =
+        QueryOuterClass.Query.newBuilder(query).setParameters(params).buildPartial();
+
+    // execute
+    return requestInfo
+        .getStargateBridge()
+        .executeQuery(finalQuery)
+        .map(
+            response -> {
+              // update next state
+              QueryOuterClass.ResultSet resultSet = response.getResultSet();
+              return resultSet;
+            });
+  }
+
+  protected static byte[] decodeBase64(String base64encoded) {
+    return Base64.getDecoder().decode(base64encoded);
+  }
+}

src/main/java/io/stargate/sgv3/docsapi/service/bridge/executor/QueryState.java

@@ -0,0 +1,39 @@
+package io.stargate.sgv3.docsapi.service.bridge.executor;
+
+import com.google.protobuf.BytesValue;
+import javax.annotation.Nullable;
+import org.immutables.value.Value;
+
+/**
+ * Utility used by the {@link QueryExecutor} in order to track the state for executing the same
+ * query in the paginated mode.
+ */
+@Value.Immutable
+public interface QueryState {
+
+  /** @return Current page size to be used with the query. */
+  @Value.Parameter
+  int pageSize();
+
+  /**
+   * @return Current paging state to be used with the query. Can be <code>null</code> to denote no
+   *     paging state exists.
+   */
+  @Value.Parameter
+  @Nullable
+  BytesValue pagingState();
+
+  /**
+   * Constructs the next state for the query.
+   *
+   * @param nextPagingState Updated paging state, usually received from the result set.
+   * @param exponentPageSize If current page size should be increased exponentially.
+   * @param maxPageSize The absolute max page size that should never be exceeded.
+   * @return Next {@link QueryState}.
+   */
+  default QueryState next(BytesValue nextPagingState, boolean exponentPageSize, int maxPageSize) {
+    int nextPageSize = exponentPageSize ? Math.min(pageSize() * 2, maxPageSize) : pageSize();
+
+    return ImmutableQueryState.of(nextPageSize, nextPagingState);
+  }
+}