ascherbakoff commented on code in PR #6007:
URL: https://github.com/apache/ignite-3/pull/6007#discussion_r2157071468
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItThinClientTransactionsTest.java:
##########
@@ -736,25 +756,145 @@ void testMixedMappingScenarioWithNoopEnlistment() throws
Exception {
for (Entry<Tuple, Tuple> entry : batch0.entrySet()) {
recs.add(kv(entry.getKey().intValue(0),
entry.getValue().stringValue(0)));
}
- assertEquals(0, recView.insertAll(tx0, recs).size()); // Proxy write
- assertEquals(recs.size(), recView.insertAll(tx0, recs).size()); //
Proxy no-op write
- assertEquals(0, recView.deleteAllExact(tx0, recs).size()); // Proxy
write
- assertEquals(recs.size(), recView.deleteAllExact(tx0, recs).size());
// Proxy no-op write
+ partitions += partitions(batch0.keySet(), table);
+ assertEquals(0, recView.insertAll(tx0, recs).size()); // Batch split
write
+ partitions += partitions(batch0.keySet(), table);
+ assertEquals(recs.size(), recView.insertAll(tx0, recs).size()); //
Batch split no-op write
+
+ partitions += partitions(batch0.keySet(), table);
+ assertEquals(0, recView.deleteAllExact(tx0, recs).size()); // Batch
split write
+ partitions += partitions(batch0.keySet(), table);
+ assertEquals(recs.size(), recView.deleteAllExact(tx0, recs).size());
// Batch split no-op write
assertTrue(recView.deleteExact(tx0, rec0)); // Write
assertFalse(recView.deleteExact(tx0, rec0)); // No-op write
tx0.commit();
// Expecting each write operation to trigger add/remove events.
- int exp = 20;
+ int exp = 20 + partitions;
Mockito.verify(spyed,
Mockito.times(exp)).addInflight(tx0.startedTx().txId());
Mockito.verify(spyed,
Mockito.times(exp)).removeInflight(Mockito.eq(tx0.startedTx().txId()),
Mockito.any());
- for (Entry<Tuple, Tuple> entry : data.entrySet()) {
- view.put(null, entry.getKey(), entry.getValue());
+ // Check if all locks are released.
+ Map<Tuple, Tuple> batch = new HashMap<>();
+
+ for (Tuple tup : tuples0) {
+ batch.put(tup, val(tup.intValue(0) + ""));
+ }
+
+ for (Tuple tup : tuples1) {
+ batch.put(tup, val(tup.intValue(0) + ""));
}
+
+ view.putAll(null, batch);
+ }
+
+ @Test
+ void testBatchScenarioWithNoopEnlistmentImplicit() {
+ Map<Partition, ClusterNode> map =
table().partitionManager().primaryReplicasAsync().join();
+
+ ClientTable table = (ClientTable) table();
+
+ IgniteImpl server0 = TestWrappers.unwrapIgniteImpl(server(0));
+ IgniteImpl server1 = TestWrappers.unwrapIgniteImpl(server(1));
+
+ List<Tuple> tuples0 = generateKeysForNode(600, 50, map,
server0.clusterService().topologyService().localMember(), table);
+ List<Tuple> tuples1 = generateKeysForNode(610, 50, map,
server1.clusterService().topologyService().localMember(), table);
+
+ Map<Tuple, Tuple> batch = new HashMap<>();
+
+ for (Tuple tup : tuples0) {
+ batch.put(tup, val(tup.intValue(0) + ""));
+ }
+
+ for (Tuple tup : tuples1) {
+ batch.put(tup, val(tup.intValue(0) + ""));
+ }
+
+ KeyValueView<Tuple, Tuple> view = table.keyValueView();
+ view.putAll(null, batch);
+
+ assertEquals(batch.size(), view.getAll(null, batch.keySet()).size());
+
+ assertEquals(0, view.removeAll(null, batch.keySet()).size());
+ assertEquals(batch.size(), view.removeAll(null,
batch.keySet()).size());
+ }
+
+ @Test
+ void testBatchScenarioWithNoopEnlistmentExplicit() {
Review Comment:
🆗
##########
modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTable.java:
##########
@@ -877,12 +886,148 @@ int tryGetPartitionCount() {
return partitionCount;
}
+ /**
+ * Batch with indexes.
+ *
+ * @param <E> Batch type element.
+ */
+ static class Batch<E> {
+ List<E> batch = new ArrayList<>();
+ List<Integer> originalIndices = new ArrayList<>();
+
+ void add(E entry, int origIdx) {
+ batch.add(entry);
+ originalIndices.add(origIdx);
+ }
+ }
+
+ private static <E> void reduceWithKeepOrder(List<E> agg, List<E> cur,
List<Integer> originalIndices) {
+ for (int i = 0; i < cur.size(); i++) {
+ E val = cur.get(i);
+ Integer orig = originalIndices.get(i);
+ agg.set(orig, val);
+ }
+ }
+
+ <R, E> CompletableFuture<R> split(
+ Transaction tx,
+ Collection<E> keys,
+ BiFunction<Collection<E>, PartitionAwarenessProvider,
CompletableFuture<R>> fun,
+ @Nullable R initialValue,
+ Reducer<R> reducer,
+ BiFunction<ClientSchema, E, Integer> hashFunc
+ ) {
+ assert tx != null;
+
+ CompletableFuture<ClientSchema> schemaFut = getSchema(latestSchemaVer);
+ CompletableFuture<List<String>> partitionsFut =
getPartitionAssignment();
+
+ return CompletableFuture.allOf(schemaFut, partitionsFut)
+ .thenCompose(v -> {
+ List<E> unmapped = new ArrayList<>();
+ Map<Integer, List<E>> mapped = new HashMap<>();
+ for (E key : keys) {
+ ClientSchema schema = schemaFut.getNow(null);
+ @Nullable List<String> aff =
partitionsFut.getNow(null);
Review Comment:
🆗
--
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-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
