You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by "tkalkirill (via GitHub)" <gi...@apache.org> on 2023/10/09 06:33:48 UTC
Re: [PR] IGNITE-20435 Preserve key order in batch opperations (deleteAll, deleteAllExact, insertAll) [ignite-3]
tkalkirill commented on code in PR #2664:
URL: https://github.com/apache/ignite-3/pull/2664#discussion_r1349896785
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java:
##########
@@ -1485,44 +1468,66 @@ public int partitionId(BinaryRowEx row) {
}
/**
- * Gathers the result of batch processing into a single resulting collection of rows.
+ * Gets a batch result.
*
* @param rowBatches Row batches.
* @return Future of collecting results.
*/
- static CompletableFuture<Collection<BinaryRow>> collectMultiRowsResponsesWithoutRestoreOrder(Collection<RowBatch> rowBatches) {
- return allResultFutures(rowBatches)
- .thenApply(response -> {
- var result = new ArrayList<BinaryRow>(rowBatches.size());
-
- for (RowBatch rowBatch : rowBatches) {
- Collection<BinaryRow> batchResult = (Collection<BinaryRow>) rowBatch.getCompletedResult();
+ public static CompletableFuture<List<BinaryRow>> collectRejectedRowsResponsesWithRestoreOrder(Collection<RowBatch> rowBatches) {
+ return collectMultiRowsResponsesWithRestoreOrder(
+ rowBatches,
+ batch -> {
+ List<BinaryRow> result = new ArrayList<>();
+ List<Boolean> response = (List<Boolean>) batch.getCompletedResult();
Review Comment:
`java.util.BitSet` will be more effective.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java:
##########
@@ -1537,7 +1540,7 @@ static CompletableFuture<List<BinaryRow>> collectMultiRowsResponsesWithRestoreOr
}
// Use Arrays#asList to avoid copying the array.
- return Arrays.asList(result);
+ return skipNull ? Arrays.stream(result).filter(r -> r != null).collect(Collectors.toList()) : Arrays.asList(result);
Review Comment:
But why wait for benchmarks?
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/UpdatableTableImpl.java:
##########
@@ -383,35 +380,26 @@ private static void appendValue(RowAssembler rowAssembler, ColumnDescriptor colD
rowAssembler.appendValue(val);
}
- private static <RowT> CompletableFuture<List<RowT>> handleInsertResults(
- RowHandler<RowT> handler,
- CompletableFuture<List<RowT>>[] futs
+ private <RowT> CompletableFuture<List<RowT>> handleInsertResults(
+ ExecutionContext<RowT> ectx,
+ Collection<RowBatch> batches
) {
- return CompletableFuture.allOf(futs)
+ return collectRejectedRowsResponsesWithRestoreOrder(batches)
.thenApply(response -> {
- List<String> conflictRows = null;
-
- for (CompletableFuture<List<RowT>> future : futs) {
- List<RowT> values = future.join();
-
- if (nullOrEmpty(values)) {
- continue;
- }
-
- if (conflictRows == null) {
- conflictRows = new ArrayList<>(values.size());
- }
-
- for (RowT row : values) {
- conflictRows.add(handler.toString(row));
- }
+ if (nullOrEmpty(response)) {
+ return null;
}
- if (conflictRows != null) {
- throw conflictKeysException(conflictRows);
- }
+ RowHandler<RowT> handler = ectx.rowHandler();
+ IgniteTypeFactory typeFactory = ectx.getTypeFactory();
+ RowSchema rowSchema = rowSchemaFromRelTypes(RelOptUtil.getFieldTypeList(desc.insertRowType(typeFactory)));
+ RowHandler.RowFactory<RowT> rowFactory = handler.factory(rowSchema);
+
+ List<String> conflictRows = response.stream()
Review Comment:
It won't take so many lines of code, why not do it more efficiently right away?
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -1734,15 +1734,17 @@ private CompletableFuture<?> processMultiEntryAction(ReadWriteMultiRowReplicaReq
return allOf(deleteExactLockFuts).thenCompose(ignore -> {
Map<UUID, BinaryRowMessage> rowIdsToDelete = new HashMap<>();
- Collection<BinaryRow> result = new ArrayList<>();
+ Collection<Boolean> result = new ArrayList<>();
Review Comment:
`java.util.BitSet` will be more effective.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -1769,7 +1771,7 @@ private CompletableFuture<?> processMultiEntryAction(ReadWriteMultiRowReplicaReq
}
return allOf(pkReadLockFuts).thenCompose(ignore -> {
- Collection<BinaryRow> result = new ArrayList<>();
+ Collection<Boolean> result = new ArrayList<>();
Review Comment:
`java.util.BitSet` will be more effective.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/RowBatch.java:
##########
@@ -31,19 +31,19 @@
*
* <p>NOTE: Not thread-safe.
*/
-class RowBatch {
+public class RowBatch {
Review Comment:
Then it needs to add encapsulation.
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/UpdatableTableImpl.java:
##########
@@ -249,59 +252,53 @@ public <RowT> CompletableFuture<?> insertAll(
RowHandler<RowT> handler = ectx.rowHandler();
- Int2ObjectOpenHashMap<List<BinaryRow>> rowsByPartition = new Int2ObjectOpenHashMap<>();
-
- for (RowT row : rows) {
- BinaryRowEx binaryRow = convertRow(row, ectx.rowHandler());
-
- rowsByPartition.computeIfAbsent(partitionExtractor.fromRow(binaryRow), k -> new ArrayList<>()).add(binaryRow);
- }
+ Int2ObjectMap<RowBatch> rowBatchByPartitionId = toRowBatchByPartitionId(handler, rows);
- CompletableFuture<List<RowT>>[] futures = new CompletableFuture[rowsByPartition.size()];
-
- int batchNum = 0;
-
- IgniteTypeFactory typeFactory = ectx.getTypeFactory();
- RowSchema rowSchema = rowSchemaFromRelTypes(RelOptUtil.getFieldTypeList(desc.insertRowType(typeFactory)));
- RowHandler.RowFactory<RowT> rowFactory = handler.factory(rowSchema);
+ for (Int2ObjectMap.Entry<RowBatch> partitionRowBatch : rowBatchByPartitionId.int2ObjectEntrySet()) {
+ int partitionId = partitionRowBatch.getIntKey();
+ RowBatch rowBatch = partitionRowBatch.getValue();
- for (Int2ObjectMap.Entry<List<BinaryRow>> partToRows : rowsByPartition.int2ObjectEntrySet()) {
- TablePartitionId partGroupId = new TablePartitionId(tableId, partToRows.getIntKey());
+ TablePartitionId partGroupId = new TablePartitionId(tableId, partitionId);
ColocationGroup group = ectx.group(MODIFY_NODE_SOURCE_ID);
assert group != null;
- NodeWithTerm nodeWithTerm = group.assignments().get(partToRows.getIntKey());
+ NodeWithTerm nodeWithTerm = group.assignments().get(partitionId);
ReplicaRequest request = MESSAGES_FACTORY.readWriteMultiRowReplicaRequest()
.groupId(partGroupId)
.commitPartitionId(serializeTablePartitionId(commitPartitionId))
- .binaryRowMessages(serializeBinaryRows(partToRows.getValue()))
+ .binaryRowMessages(serializeBinaryRows(rowBatch.requestedRows))
.transactionId(txAttributes.id())
.term(nodeWithTerm.term())
.requestType(RequestType.RW_INSERT_ALL)
.timestampLong(clock.nowLong())
.build();
- futures[batchNum++] = replicaService.invoke(nodeWithTerm.name(), request)
- .thenApply(result -> {
- Collection<BinaryRow> binaryRows = (Collection<BinaryRow>) result;
+ rowBatch.resultFuture = replicaService.invoke(nodeWithTerm.name(), request);
+ }
+
+ return handleInsertResults(ectx, rowBatchByPartitionId.values());
+ }
- if (binaryRows.isEmpty()) {
- return List.of();
- }
+ /**
+ * Creates batches of rows for processing, grouped by partition ID.
+ *
+ * @param rowHandler Row handler.
+ * @param rows Rows.
+ */
+ private <T> Int2ObjectMap<RowBatch> toRowBatchByPartitionId(RowHandler<T> rowHandler, List<T> rows) {
+ Int2ObjectMap<RowBatch> rowBatchByPartitionId = new Int2ObjectOpenHashMap<>();
- List<RowT> conflictRows = new ArrayList<>(binaryRows.size());
+ int i = 0;
- for (BinaryRow row : binaryRows) {
- conflictRows.add(rowConverter.toRow(ectx, row, rowFactory));
- }
+ for (T row : rows) {
Review Comment:
In your case, it will be more effective, one less allocation plus various kinds of optimization.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org