Is there a working example of the Quarkus Cassandra Client for Pagination?

Question

I am still lost using the Quarkus Cassandra client https://quarkus.io/guides/cassandra I am trying the whole day to get a pagination done with the quarkus integration but the documentation is not realy helpful at all. Somebody else here has a working example for pagination with the Quarkus Cassandra client? Please share your knowledge with me. Thanks in advance.

EDIT: This my sample project/playground: https://github.com/edward-fakhouri/quarkus-cassandra-playground Here you can find a working example.

I have found something in this cassandra documentation: https://docs.datastax.com/en/developer/java-driver/3.1/manual/paging/ and implemented OffsetPagination but it isn't realy what I am searching for...

Answer 1

There is pagination in the Quarkus Cassandra client however I can't make it work as I would like either. Below you see an example to get the pagination state with the example from the quarkus documentation https://quarkus.io/guides/cassandra

public PagingFruit getPagingState(String pagingState){
        ByteBuffer state = ByteBuffer.wrap(pagingState.getBytes());
        MutinyMappedReactiveResultSet<Fruit> fruit = this.fruitMapper.fruitDao().findAll(x -> x.setPagingState(state).setPageSize(1000); // here we provide the unary operator that sets the state.  
        var newState = Uni.createFrom().publisher(fruit.getExecutionInfos()).map(ExecutionInfo::getPagingState).await().indefinitely().toString();
        return new PagingFruit(fruit.collect().asList().await().indefinitely(), newState);
}

record PagingFruit(List<Fruit> fruit, String pagingState){}

This uses the MutinyMappedReactiveResultSet which allows you to query the ExecutionInfos.

The problem I have at this moment using the quarkus implementation is that I need to query Cassandra twice to actually provide the data and the pagination with it. Therefore I opted for a prepared statement like this example:

class PaginatedRepository {

    private final QuarkusCqlSession cqlSession;
    private final PreparedStatement query;
    
    PaginatedRepository(QuarkusCqlSession quarkusCqlSession){
        this.cqlSession = quarkusCqlSession;
        this.query = cqlSession.prepare("SELECT fruit FROM keypace.table");
    }
    
    public Uni<PagingFruit> findFruit(String page) {
        var statement = query
                .bind()
                .setPageSize(1000)
                .setPagingState(PagingState.fromString(page));
    
        return Uni.createFrom().completionStage(cqlSession.executeAsync(statement))
                .map(resultSet -> {
                    var pagingState = resultSet.getExecutionInfo().getSafePagingState();
                    var newCursor = (pagingState == null) ? null : pagingState.toString();
                    var rows = Streams.stream(resultSet.currentPage())
                            .map(row -> row.get(1, Fruit.class)
                            ).filter(Objects::nonNull).toList();
                    return new PagingFruit(rows, newCursor);
                });
    }

This is untested code, just the general concept which should work.

Answer 2

So, in my oppinion and my research, there is no convinient way to do pagination on a cassandra database with the quarkus cassandra client. You have to handle it by yourself.

Answer 3

With the latest cassandra bom you can run those tests

<dependency>
    <groupId>com.datastax.oss.quarkus</groupId>
    <artifactId>cassandra-quarkus-bom</artifactId>
    <version>1.1.2</version>
    <type>pom</type>
    <scope>import</scope>
</dependency>

with that docker cassandra started working, after that your function started working

    @Override
    public List<Customer> findPagedCustomerByCustomerNumber(String customerNumber, int pageNumber, int pageSize) {
        List<Customer> result = new ArrayList<>();
        PreparedStatement query = cqlSession.prepare(
                "SELECT * FROM test.customer WHERE customer_number = :customerNumber ORDER BY  creation_date");
        BoundStatement completeStatement = query.bind().setString("customerNumber", customerNumber);
        OffsetPager pager = new OffsetPager(pageSize);
        ResultSet resultSet = cqlSession.execute(completeStatement);
        OffsetPager.Page<Row> page = pager.getPage(resultSet, pageNumber);
        List<Row> pageElements = page.getElements();
        pageElements.forEach(c -> result.add(Customer.builder()
                .customerNumber(c.getString("customer_number"))
                .creationDate(c.getLocalDate("creation_date"))
                .description(c.getString("description"))
                .state(c.getString("state"))
                .build())
        );
        return result;
    }

I have been using querybuilder for a long time so I added that too

public List<Customer> findPagedCustomerByCustomerNumberPaging(String customerNumber, int pageNumber, int pageSize) {
        List<Customer> result = new ArrayList<>();
        Select select = QueryBuilder.selectFrom("test", "customer")
                .columns("customer_number", "creation_date", "description", "state")
                .whereColumn("customer_number").isEqualTo(literal(customerNumber))
                .orderBy("createtion_date", ClusteringOrder.ASC)
                .allowFiltering();

        SimpleStatement completeStatement = select.build();
        OffsetPager pager = new OffsetPager(pageSize);
        ResultSet resultSet = cqlSession.execute(completeStatement);
        OffsetPager.Page<Row> page = pager.getPage(resultSet, pageNumber);
        List<Row> pageElements = page.getElements();
        pageElements.forEach(c -> result.add(Customer.builder()
                .customerNumber(c.getString("customer_number"))
                .creationDate(c.getLocalDate("creation_date"))
                .description(c.getString("description"))
                .state(c.getString("state"))
                .build())
        );
        return result;
    }

I understand cassandra-quarkus-test-framework brings docker but in case you dont have docker you can always use CassandraUnit which is lighter then docker. All code can be found here.