fixes pagination

Author: otaviojava

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

@@ -23,6 +23,9 @@
 interface QueryExecutor {
 
     static QueryExecutor of(ColumnQuery query) {
+        if (CassandraQuery.class.isInstance(query)) {
+            return QueryExecutorType.PAGING_STATE;
+        }
         return QueryExecutorType.DEFAULT;
     }
 

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

@@ -18,6 +18,7 @@
 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.core.paging.OffsetPager;
 import com.datastax.oss.driver.api.querybuilder.select.Select;
 import jakarta.nosql.column.ColumnEntity;
 import jakarta.nosql.column.ColumnQuery;
@@ -81,11 +82,22 @@ public Stream<ColumnEntity> execute(String keyspace, ColumnQuery query, DefaultC
         @Override
         public Stream<ColumnEntity> execute(String keyspace, ColumnQuery query, ConsistencyLevel level,
                                             DefaultCassandraColumnFamilyManager manager) {
-            SimpleStatement select = QueryUtils.select(query, keyspace).build();
+
+            final Select cassandraSelect = QueryUtils.select(query, keyspace);
+
+
+            if (query.getLimit() > 0 && query.getSkip() == 0) {
+                cassandraSelect.limit((int) query.getLimit());
+            }
+
+            SimpleStatement select = cassandraSelect.build();
             if (Objects.nonNull(level)) {
                 select = select.setConsistencyLevel(level);
             }
             ResultSet resultSet = manager.getSession().execute(select);
+            if (query.getLimit() > 0 && query.getSkip() > 0) {
+                return resultSet.all().stream().skip(query.getSkip()).limit(query.getLimit()).map(CassandraConverter::toDocumentEntity);
+            }
             return resultSet.all().stream().map(CassandraConverter::toDocumentEntity);
         }
     }

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

@@ -85,13 +85,6 @@ public static Select select(ColumnQuery query, String keyspace) {
             select = QueryBuilder.selectFrom(keyspace, columnFamily).columns(columns);
         }
 
-        if (query.getLimit() > 0) {
-            if (CassandraQuery.class.isInstance(query)) {
-                //
-            } else {
-                select = select.limit((int) query.getLimit());
-            }
-        }
         select = select.where(Relations.createClause(query.getCondition().orElse(null)));
         final Map<String, ClusteringOrder> sort = query.getSorts().stream().collect(Collectors.toMap(s -> s.getName(), mapSort()));
         select = select.orderBy(sort);

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

@@ -478,33 +478,20 @@ public void shouldCount() {
 
    @Test
     public void shouldPagingState() {
-        for (long index = 1; index < 10; index++) {
+        for (long index = 1; index <= 10; index++) {
             ColumnEntity columnFamily = getColumnFamily();
             columnFamily.add("id", index);
             entityManager.insert(columnFamily);
         }
 
-        ColumnQuery query = select().from(Constants.COLUMN_FAMILY).limit(6).build();
+        ColumnQuery query = select().from(Constants.COLUMN_FAMILY).build();
         CassandraQuery cassandraQuery = CassandraQuery.of(query);
 
         assertFalse(cassandraQuery.getPagingState().isPresent());
 
         List<ColumnEntity> entities = entityManager.select(cassandraQuery).collect(toList());
-        assertEquals(6, entities.size());
-        assertTrue(cassandraQuery.getPagingState().isPresent());
-
-        entities = entityManager.select(cassandraQuery).collect(toList());
-        assertEquals(3, entities.size());
-        assertTrue(cassandraQuery.getPagingState().isPresent());
-
-        entities = entityManager.select(cassandraQuery).collect(toList());
-        assertTrue(entities.isEmpty());
+        assertEquals(10, entities.size());
         assertTrue(cassandraQuery.getPagingState().isPresent());
-
-        entities = entityManager.select(cassandraQuery).collect(toList());
-        assertTrue(entities.isEmpty());
-        assertTrue(cassandraQuery.getPagingState().isPresent());
-
     }
 
     private ColumnEntity createEntityWithIterable() {