creates paginate

Author: otaviojava

cassandra-driver/src/main/java/org/eclipse/jnosql/diana/cassandra/column/CassandraQuery.java

@@ -20,6 +20,8 @@
 import jakarta.nosql.column.ColumnCondition;
 import jakarta.nosql.column.ColumnQuery;
 
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
@@ -60,6 +62,16 @@ public Optional<String> getPagingState() {
         return Optional.ofNullable(pagingState);
     }
 
+    Optional<ByteBuffer> toPaginate() {
+        return getPagingState().filter(NOT_EQUALS).map(s -> ByteBuffer.wrap(s.getBytes(StandardCharsets.UTF_8)));
+    }
+
+    void setPagingState(ByteBuffer pagingState) {
+        if (pagingState != null) {
+            this.pagingState = StandardCharsets.UTF_8.decode(pagingState).toString();
+        }
+    }
+
 
     void setExhausted(boolean exhausted) {
         synchronized (this) {

cassandra-driver/src/main/java/org/eclipse/jnosql/diana/cassandra/column/QueryExecutorType.java

@@ -16,15 +16,62 @@
 
 import com.datastax.oss.driver.api.core.ConsistencyLevel;
 import com.datastax.oss.driver.api.core.cql.ResultSet;
+import com.datastax.oss.driver.api.core.cql.Row;
 import com.datastax.oss.driver.api.core.cql.SimpleStatement;
+import com.datastax.oss.driver.api.querybuilder.select.Select;
 import jakarta.nosql.column.ColumnEntity;
 import jakarta.nosql.column.ColumnQuery;
 
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Objects;
 import java.util.stream.Stream;
 
 enum QueryExecutorType implements QueryExecutor {
+    PAGING_STATE {
+        @Override
+        public Stream<ColumnEntity> execute(String keyspace, ColumnQuery query, DefaultCassandraColumnFamilyManager manager) {
+            return execute(keyspace, query, null, manager);
+        }
+
+        @Override
+        public Stream<ColumnEntity> execute(String keyspace, ColumnQuery q, ConsistencyLevel level,
+                                            DefaultCassandraColumnFamilyManager manager) {
+
+            CassandraQuery query = CassandraQuery.class.cast(q);
+
+            if (query.isExhausted()) {
+                return Stream.empty();
+            }
+            Select select = QueryUtils.select(query, keyspace);
+            SimpleStatement simpleStatement = select.build();
+            if (Objects.nonNull(level)) {
+                simpleStatement = simpleStatement.setConsistencyLevel(level);
+            }
+
+            if (query.toPaginate().isPresent()) {
+                simpleStatement = simpleStatement.setPagingState(query.toPaginate().get());
+            }
+
+            ResultSet resultSet = manager.getSession().execute(simpleStatement);
+
+            final ByteBuffer pagingState = resultSet.getExecutionInfo().getPagingState();
+            query.setPagingState(pagingState);
+
+            List<ColumnEntity> entities = new ArrayList<>();
+
+            for (Row row : resultSet) {
+                entities.add(CassandraConverter.toDocumentEntity(row));
+                if (resultSet.getAvailableWithoutFetching() == 0) {
+                    query.setExhausted(resultSet.isFullyFetched());
+                    break;
+                }
+            }
+            return entities.stream();
+        }
 
+    },
     DEFAULT {
         @Override
         public Stream<ColumnEntity> execute(String keyspace, ColumnQuery query, DefaultCassandraColumnFamilyManager manager) {

cassandra-driver/src/test/java/org/eclipse/jnosql/diana/cassandra/column/CassandraColumnFamilyManagerTest.java

@@ -476,7 +476,7 @@ public void shouldCount() {
         assertTrue(contacts > 0);
     }
 
-   /* @Test
+   @Test
     public void shouldPagingState() {
         for (long index = 1; index < 10; index++) {
             ColumnEntity columnFamily = getColumnFamily();
@@ -505,7 +505,7 @@ public void shouldPagingState() {
         assertTrue(entities.isEmpty());
         assertTrue(cassandraQuery.getPagingState().isPresent());
 
-    }*/
+    }
 
     private ColumnEntity createEntityWithIterable() {
         ColumnEntity entity = ColumnEntity.of("contacts");