[Native Protocol] Read and count operation

src/main/java/io/stargate/sgv2/jsonapi/service/bridge/executor/QueryExecutor.java

@@ -2,7 +2,6 @@
 
 import com.datastax.oss.driver.api.core.ConsistencyLevel;
 import com.datastax.oss.driver.api.core.cql.AsyncResultSet;
-import com.datastax.oss.driver.api.core.cql.BoundStatement;
 import com.datastax.oss.driver.api.core.cql.SimpleStatement;
 import com.datastax.oss.driver.api.core.metadata.schema.KeyspaceMetadata;
 import com.datastax.oss.driver.api.core.metadata.schema.TableMetadata;
@@ -69,24 +68,24 @@ public Uni<QueryOuterClass.ResultSet> executeRead(
   /**
    * Execute read query with bound statement.
    *
-   * @param boundStatement - Bound statement with query and parameters. The table name used in the
+   * @param simpleStatement - Simple statement with query and parameters. The table name used in the
    *     query must have keyspace prefixed.
    * @param pagingState - In case of pagination, the paging state needs to be passed to fetch
    *     subsequent pages
    * @param pageSize - page size
    * @return AsyncResultSet
    */
   public Uni<AsyncResultSet> executeRead(
-      BoundStatement boundStatement, Optional<String> pagingState, int pageSize) {
+      SimpleStatement simpleStatement, Optional<String> pagingState, int pageSize) {
     if (pagingState.isPresent()) {
-      boundStatement =
-          boundStatement
+      simpleStatement =
+          simpleStatement
               .setSerialConsistencyLevel(getConsistencyLevel(queriesConfig.consistency().reads()))
               .setPageSize(pageSize)
               .setPagingState(ByteBuffer.wrap(decodeBase64(pagingState.get())));
     }
     return Uni.createFrom()
-        .completionStage(cqlSessionCache.getSession().executeAsync(boundStatement));
+        .completionStage(cqlSessionCache.getSession().executeAsync(simpleStatement));
   }
 
   /**
@@ -113,17 +112,17 @@ public Uni<QueryOuterClass.ResultSet> executeWrite(QueryOuterClass.Query query)
   /**
    * Execute write query with bound statement.
    *
-   * @param boundStatement - Bound statement with query and parameters. The table name used in the
-   *     query must have keyspace prefixed.
+   * @param statement - Bound statement with query and parameters. The table name used in the query
+   *     must have keyspace prefixed.
    * @return AsyncResultSet
    */
-  public Uni<AsyncResultSet> executeWrite(BoundStatement boundStatement) {
+  public Uni<AsyncResultSet> executeWrite(SimpleStatement statement) {
     return Uni.createFrom()
         .completionStage(
             cqlSessionCache
                 .getSession()
                 .executeAsync(
-                    boundStatement
+                    statement
                         .setConsistencyLevel(
                             getConsistencyLevel(queriesConfig.consistency().writes()))
                         .setSerialConsistencyLevel(

src/main/java/io/stargate/sgv2/jsonapi/service/bridge/serializer/CQLBindValues.java

@@ -0,0 +1,91 @@
+package io.stargate.sgv2.jsonapi.service.bridge.serializer;
+
+import com.datastax.oss.driver.api.core.data.CqlVector;
+import com.datastax.oss.driver.api.core.data.TupleValue;
+import com.datastax.oss.driver.api.core.type.DataTypes;
+import com.datastax.oss.driver.api.core.type.TupleType;
+import io.stargate.sgv2.jsonapi.service.shredding.JsonPath;
+import io.stargate.sgv2.jsonapi.service.shredding.model.DocumentId;
+import java.math.BigDecimal;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class CQLBindValues {
+
+  public static Map<String, Integer> getIntegerMapValues(Map<JsonPath, Integer> from) {
+    final Map<String, Integer> to = new HashMap<>(from.size());
+    for (Map.Entry<JsonPath, Integer> entry : from.entrySet()) {
+      to.put(entry.getKey().toString(), entry.getValue());
+    }
+    return to;
+  }
+
+  public static Set<String> getSetValue(Set<JsonPath> from) {
+    return from.stream().map(val -> val.toString()).collect(Collectors.toSet());
+  }
+
+  public static Set<String> getStringSetValue(Set<String> from) {
+    return from.stream().map(val -> val.toString()).collect(Collectors.toSet());
+  }
+
+  public static List<String> getListValue(List<JsonPath> from) {
+    return from.stream().map(val -> val.toString()).collect(Collectors.toList());
+  }
+
+  public static Map<String, String> getStringMapValues(Map<JsonPath, String> from) {
+    final Map<String, String> to = new HashMap<>(from.size());
+    for (Map.Entry<JsonPath, String> entry : from.entrySet()) {
+      to.put(entry.getKey().toString(), entry.getValue());
+    }
+    return to;
+  }
+
+  public static Map<String, Byte> getBooleanMapValues(Map<JsonPath, Boolean> from) {
+    final Map<String, Byte> to = new HashMap<>(from.size());
+    for (Map.Entry<JsonPath, Boolean> entry : from.entrySet()) {
+      to.put(entry.getKey().toString(), (byte) (entry.getValue() ? 1 : 0));
+    }
+    return to;
+  }
+
+  public static Map<String, BigDecimal> getDoubleMapValues(Map<JsonPath, BigDecimal> from) {
+    final Map<String, BigDecimal> to = new HashMap<>(from.size());
+    for (Map.Entry<JsonPath, BigDecimal> entry : from.entrySet()) {
+      to.put(entry.getKey().toString(), entry.getValue());
+    }
+    return to;
+  }
+
+  public static Map<String, Instant> getTimestampMapValues(Map<JsonPath, Date> from) {
+    final Map<String, Instant> to = new HashMap<>(from.size());
+    for (Map.Entry<JsonPath, Date> entry : from.entrySet()) {
+      to.put(entry.getKey().toString(), Instant.ofEpochMilli(entry.getValue().getTime()));
+    }
+    return to;
+  }
+
+  private static TupleType tupleType = DataTypes.tupleOf(DataTypes.TINYINT, DataTypes.TEXT);
+
+  public static TupleValue getDocumentIdValue(DocumentId documentId) {
+    // Temporary implementation until we convert it to Tuple in DB
+    final TupleValue tupleValue =
+        tupleType.newValue((byte) documentId.typeId(), documentId.asDBKey());
+    return tupleValue;
+  }
+
+  public static CqlVector<Float> getVectorValue(float[] vectors) {
+    if (vectors == null || vectors.length == 0) {
+      return null;
+    }
+
+    List<Float> vectorValues = new ArrayList<>(vectors.length);
+    for (float vectorValue : vectors) vectorValues.add(vectorValue);
+    return CqlVector.newInstance(vectorValues);
+  }
+}

src/main/java/io/stargate/sgv2/jsonapi/service/operation/model/CountOperation.java

@@ -1,6 +1,7 @@
 package io.stargate.sgv2.jsonapi.service.operation.model;
 
 import com.bpodgursky.jbool_expressions.Expression;
+import com.datastax.oss.driver.api.core.cql.SimpleStatement;
 import io.smallrye.mutiny.Uni;
 import io.stargate.bridge.proto.QueryOuterClass;
 import io.stargate.sgv2.api.common.cql.builder.BuiltCondition;
@@ -11,8 +12,12 @@
 import io.stargate.sgv2.jsonapi.service.bridge.executor.QueryExecutor;
 import io.stargate.sgv2.jsonapi.service.operation.model.impl.CountOperationPage;
 import io.stargate.sgv2.jsonapi.service.operation.model.impl.ExpressionBuilder;
+import io.stargate.sgv2.jsonapi.service.operation.model.impl.JsonTerm;
+import java.util.LinkedHashSet;
 import java.util.List;
+import java.util.Set;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
 
 /**
  * Operation that returns count of documents based on the filter condition. Written with the
@@ -23,21 +28,31 @@ public record CountOperation(CommandContext commandContext, LogicalExpression lo
 
   @Override
   public Uni<Supplier<CommandResult>> execute(QueryExecutor queryExecutor) {
-    QueryOuterClass.Query query = buildSelectQuery();
-    return countDocuments(queryExecutor, query)
+    SimpleStatement simpleStatement = buildSelectQuery();
+    return countDocuments(queryExecutor, simpleStatement)
         .onItem()
         .transform(docs -> new CountOperationPage(docs.count()));
   }
 
-  private QueryOuterClass.Query buildSelectQuery() {
+  private SimpleStatement buildSelectQuery() {
     List<Expression<BuiltCondition>> expressions =
         ExpressionBuilder.buildExpressions(logicalExpression, null);
-    return new QueryBuilder()
-        .select()
-        .count()
-        .as("count")
-        .from(commandContext.namespace(), commandContext.collection())
-        .where(expressions.get(0)) // TODO count will assume no id filter query split?
-        .build();
+    Set<BuiltCondition> conditions = new LinkedHashSet<>();
+    expressions.get(0).collectK(conditions, Integer.MAX_VALUE);
+    final List<Object> collect =
+        conditions.stream()
+            .map(builtCondition -> ((JsonTerm) builtCondition.value()).get())
+            .collect(Collectors.toList());
+    final QueryOuterClass.Query query =
+        new QueryBuilder()
+            .select()
+            .count()
+            .as("count")
+            .from(commandContext.namespace(), commandContext.collection())
+            .where(expressions.get(0)) // TODO count will assume no id filter query split?
+            .build();
+
+    final SimpleStatement simpleStatement = SimpleStatement.newInstance(query.getCql());
+    return simpleStatement.setPositionalValues(collect);
   }
 }