tkalkirill commented on code in PR #2236:
URL: https://github.com/apache/ignite-3/pull/2236#discussion_r1239494792
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java:
##########
@@ -1206,28 +1208,87 @@ public int partitionId(BinaryRowEx row) {
}
/**
- * TODO asch keep the same order as for keys Collects multirow responses
from multiple futures into a single collection IGNITE-16004.
+ * Gathers the result of batch processing into a single resulting
collection of rows.
*
- * @param futs Futures.
- * @return Row collection.
+ * @param rowBatchByPartitionId Row batches by partition ID.
+ * @return Future of collecting results.
*/
- private CompletableFuture<Collection<BinaryRow>>
collectMultiRowsResponses(CompletableFuture<Object>[] futs) {
- return CompletableFuture.allOf(futs)
+ static CompletableFuture<Collection<BinaryRow>>
collectMultiRowsResponsesWithoutRestoreOrder(
+ Int2ObjectMap<RowBatch> rowBatchByPartitionId
+ ) {
+ return allOfResultFuture(rowBatchByPartitionId)
.thenApply(response -> {
- Collection<BinaryRow> list = new ArrayList<>(futs.length);
+ var result = new
ArrayList<BinaryRow>(rowBatchByPartitionId.size());
- for (CompletableFuture<Object> future : futs) {
- Collection<BinaryRow> values = (Collection<BinaryRow>)
future.join();
+ for (Int2ObjectMap.Entry<RowBatch> partitionRowBatch :
rowBatchByPartitionId.int2ObjectEntrySet()) {
+ Collection<BinaryRow> batchResult =
getBatchProcessingResult(partitionRowBatch);
- if (values != null) {
- list.addAll(values);
+ if (batchResult == null) {
+ continue;
}
+
+ result.addAll(batchResult);
}
- return list;
+ return result;
});
}
+ /**
+ * Gathers the result of batch processing into a single resulting
collection of rows, restoring order as in the requested collection of
+ * rows.
+ *
+ * @param rowBatchByPartitionId Row batches by partition ID.
+ * @return Future of collecting results.
+ */
+ static CompletableFuture<Collection<BinaryRow>>
collectMultiRowsResponsesWithRestoreOrder(
+ Int2ObjectMap<RowBatch> rowBatchByPartitionId
+ ) {
+ return allOfResultFuture(rowBatchByPartitionId)
+ .thenApply(response -> {
+ int totalRowSize =
rowBatchByPartitionId.values().stream().mapToInt(value ->
value.rows.size()).sum();
+
+ var result = new BinaryRow[totalRowSize];
+
+ for (Int2ObjectMap.Entry<RowBatch> partitionRowBatch :
rowBatchByPartitionId.int2ObjectEntrySet()) {
+ Collection<BinaryRow> batchResult =
getBatchProcessingResult(partitionRowBatch);
+
+ if (batchResult == null) {
+ continue;
+ }
+
+ RowBatch rowBatch = partitionRowBatch.getValue();
+
+ assert batchResult.size() == rowBatch.rows.size() :
+ "batchResult=" + batchResult.size() + ",
rowBatch=" + rowBatch.rows.size();
+
+ int i = 0;
+
+ for (BinaryRow resultRow : batchResult) {
+ result[rowBatch.originalRowOrder.getInt(i++)] =
resultRow;
+ }
+ }
+
+ // Use Arrays#asList to avoid copying the array.
+ return Arrays.asList(result);
+ });
+ }
+
+ private static CompletableFuture<Void>
allOfResultFuture(Int2ObjectMap<RowBatch> rowBatchByPartitionId) {
+ return CompletableFuture.allOf(
+ rowBatchByPartitionId.values().stream().map(rowBatch ->
rowBatch.resultFuture).toArray(CompletableFuture[]::new)
+ );
+ }
+
+ private static Collection<BinaryRow>
getBatchProcessingResult(Int2ObjectMap.Entry<RowBatch> partitionRowBatch) {
+ RowBatch rowBatch = partitionRowBatch.getValue();
+
+ assert rowBatch.resultFuture != null : "partitionId=" +
partitionRowBatch.getIntKey();
+
+ // It is safe to call join(), we have already waited for all the
resultFutures.
+ return (Collection<BinaryRow>) rowBatch.resultFuture.join();
Review Comment:
Fix it
--
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]
