Skip to content

Fix for countDocument regression #1711

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
maheshrajamani wants to merge 3 commits into
base: main
Choose a base branch
from
Draft

Fix for countDocument regression #1711

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b587df6
Fix for count
maheshrajamani Nov 13, 2024
d976015
Merge branch 'main' into count-bug-fix
tatu-at-datastax Nov 13, 2024
69fc708
Merge branch 'main' into count-bug-fix
tatu-at-datastax Jan 31, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 30 additions & 13 deletions .../java/io/stargate/sgv2/jsonapi/service/operation/collections/CollectionReadOperation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
package io.stargate.sgv2.jsonapi.service.operation.collections;

import com.datastax.oss.driver.api.core.AsyncPagingIterable;
import com.datastax.oss.driver.api.core.cql.AsyncResultSet;
import com.datastax.oss.driver.api.core.cql.Row;
import com.datastax.oss.driver.api.core.cql.SimpleStatement;
Expand Down Expand Up @@ -389,21 +390,37 @@ default Uni<CountResponse> countDocumentsByKey(
QueryExecutor queryExecutor,
SimpleStatement simpleStatement) {
AtomicLong counter = new AtomicLong();
final CompletionStage<AsyncResultSet> async =
queryExecutor
.executeCount(dataApiRequestInfo, simpleStatement)
.whenComplete(
(rs, error) -> {
getCount(rs, error, counter);
});

return Uni.createFrom()
.completionStage(async)
return Multi.createBy()
.repeating()
.uni(
() -> new AtomicReference<AsyncResultSet>(null), // the state passed to the producer
stateRef -> {
Uni<AsyncResultSet> result =
stateRef.get() == null
? Uni.createFrom()
.completionStage(
queryExecutor.executeCount(
dataApiRequestInfo,
simpleStatement)) // AsyncResultSet is null so first page
: Uni.createFrom()
.completionStage(stateRef.get().fetchNextPage()); // After first page
// returning result for looping, this is remembering the rows as we page through them
return result
.onItem()
.invoke(
rs -> {
counter.addAndGet(rs.remaining());
stateRef.set(rs);
});
})
// Documents read until pageState available, max records read is deleteLimit + 1 TODO
// COMMENTS
.whilst(AsyncPagingIterable::hasMorePages)
Copy link
Contributor

@tatu-at-datastax tatu-at-datastax Nov 13, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Quick question: where is the "max count" checked? Looks like this would read through all entries and not stop at specific count? (I am probably missing something)

Copy link
Contributor

@amorton amorton Nov 14, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this would be set in the CQL limit, the page size we tell the driver is how to chunk LIMIT number of rows.

Copy link
Contributor

@amorton amorton Nov 14, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this looks like the code we came up with for in memory sorting for tables

Copy link
Contributor

@tatu-at-datastax tatu-at-datastax Nov 14, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah. Yes, since that's within SimpleStatement being passed.

.collect()
.asList()
.onItem()
.transform(
rs -> {
return new CountResponse(counter.get());
});
.transformToUni(resultSets -> Uni.createFrom().item(new CountResponse(counter.get())));
}

/**
Expand Down