sanpwc commented on a change in pull request #348:
URL: https://github.com/apache/ignite-3/pull/348#discussion_r716912174
##########
File path:
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
##########
@@ -302,4 +336,146 @@ private int partId(BinaryRow row) {
return list;
});
}
+
+ /** Partition scan publisher. */
+ private class PartitionScanPublisher implements Publisher<BinaryRow> {
+ /** {@link Publisher<BinaryRow>} that relatively notifies about
partition rows. */
+ private final RaftGroupService raftGrpSvc;
+
+ /** */
+ private AtomicBoolean wasSubscribed;
+
+ /**
+ * The constructor.
+ *
+ * @param raftGrpSvc {@link RaftGroupService} to run corresponding
raft commands.
+ */
+ PartitionScanPublisher(RaftGroupService raftGrpSvc) {
+ this.raftGrpSvc = raftGrpSvc;
+ this.wasSubscribed = new AtomicBoolean(false);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void subscribe(Subscriber<? super BinaryRow>
subscriber) {
+ if (subscriber == null)
+ throw new NullPointerException("Subscriber is null");
+
+ if (!wasSubscribed.compareAndSet(false, true))
+ subscriber.onError(new IllegalStateException("Scan publisher
does not support multiple subscriptions."));
+
+ PartitionScanSubscription subscription = new
PartitionScanSubscription(subscriber);
+
+ subscriber.onSubscribe(subscription);
+ }
+
+ /**
+ * Partition Scan Subscription.
+ */
+ private class PartitionScanSubscription implements Subscription {
+ /** */
+ private final Subscriber<? super BinaryRow> subscriber;
+
+ /** */
+ private final AtomicBoolean isCanceled;
+
+ /** Scan id to uniquely identify it on server side. */
+ private final IgniteUuid scanId;
+
+ /** Scan initial operation that created server cursor. */
+ private final CompletableFuture<Void> scanInitOp;
+
+ /**
+ * The constructor.
+ * @param subscriber The subscriber.
+ */
+ private PartitionScanSubscription(Subscriber<? super BinaryRow>
subscriber) {
+ this.subscriber = subscriber;
+ this.isCanceled = new AtomicBoolean(false);
+ this.scanId = UUID_GENERATOR.randomUuid();
+ // TODO: IGNITE-15544 Close partition scans on node left.
+ this.scanInitOp = raftGrpSvc.run(new ScanInitCommand("",
scanId));
+ }
+
+ /** {@inheritDoc} */
+ @Override public void request(long n) {
+ if (n <= 0) {
+ cancel();
+
+ subscriber.onError(new
IllegalArgumentException(LoggerMessageHelper.
+ format("Invalid requested amount of items
[requested={}, minValue=1]", n))
+ );
+ }
+
+ if (isCanceled.get())
+ return;
+
+ final int internalBatchSize = Integer.MAX_VALUE;
+
+ for (int intBatchCnr = 0; intBatchCnr < (n /
internalBatchSize); intBatchCnr++)
+ scanBatch(internalBatchSize);
+
+ scanBatch((int)(n % internalBatchSize));
+ }
+
+ /** {@inheritDoc} */
+ @Override public void cancel() {
+ cancel(true);
+ }
+
+ /**
+ * Cancels given subscription and closes cursor if necessary.
+ *
+ * @param closeCursor If {@code true} closes inner storage scan.
+ */
+ private void cancel(boolean closeCursor) {
+ isCanceled.set(true);
+
+ if (closeCursor) {
+ raftGrpSvc.run(new
ScanCloseCommand(scanId)).exceptionally(closeT -> {
+ LOG.warn("Unable to close scan.", closeT);
+
+ return null;
+ });
+ }
+ }
+
+ /**
+ * Requests and processes n requested elements where n is an
integer.
+ *
+ * @param n Requested amount of items.
+ */
+ private void scanBatch(int n) {
+ if (isCanceled.get())
+ return;
+
+ scanInitOp.thenCompose((none) ->
raftGrpSvc.<MultiRowsResponse>run(new ScanRetrieveBatchCommand(n, scanId)))
Review comment:
> We need to check isCanceled flag right before invocation of raftGrpSvc
.run(ScanRetrieveBatchCommand)
Without extra synchronization that won't help.
However, I believe it's not a big deal either. According to
`Subscription#cancel`, cancel is a sort of eventually stuff
```
Causes the Subscriber to (eventually) stop receiving messages.
Implementation is best-effort -- additional messages may be received after
invoking this method. A cancelled subscription need not ever receive an
onComplete or onError signal.
```
So in order to be complaint with Flow contract, onError() should be
prevented in case `SCAN_INIT`, `SCAN_CLOSE`, `SCAN_RETRIEVE`. I mean that if
`SCAN_CLOSE` removes the cursor and `SCAN_RETRIEVE` returns
`NoSuchElementException` we shouldn't fire onError, following code was added to
exceptionally
```
.exceptionally(
t -> {
if (t instanceof NoSuchElementException ||
t instanceof CompletionException &&
t.getCause() instanceof NoSuchElementException)
return null;
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
