tkalkirill commented on code in PR #2236:
URL: https://github.com/apache/ignite-3/pull/2236#discussion_r1239494442
##########
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();
Review Comment:
Fix it
##########
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() :
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]
