tkalkirill commented on code in PR #2664:
URL: https://github.com/apache/ignite-3/pull/2664#discussion_r1348301862
##########
modules/client/src/test/java/org/apache/ignite/client/fakes/FakeInternalTable.java:
##########
@@ -220,7 +220,7 @@ public CompletableFuture<Boolean> insert(BinaryRowEx row,
@Nullable InternalTran
/** {@inheritDoc} */
@Override
- public CompletableFuture<Collection<BinaryRow>>
insertAll(Collection<BinaryRowEx> rows, @Nullable InternalTransaction tx) {
+ public CompletableFuture<List<BinaryRow>>
insertAll(Collection<BinaryRowEx> rows, @Nullable InternalTransaction tx) {
Review Comment:
Your task is only about deleting, not about inserting data.
##########
modules/client/src/test/java/org/apache/ignite/client/fakes/FakeInternalTable.java:
##########
@@ -333,7 +333,7 @@ public CompletableFuture<BinaryRow>
getAndDelete(BinaryRowEx row, @Nullable Inte
/** {@inheritDoc} */
Review Comment:
```suggestion
```
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItInternalTableTest.java:
##########
@@ -384,78 +359,190 @@ public void testRoScanAllExplicitPopulatingData() throws
InterruptedException {
roScanAll(false);
}
- private void roScanAll(boolean implicit) throws InterruptedException {
- IgniteImpl node = node();
+ @Test
+ public void getAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
- KeyValueView<Tuple, Tuple> keyValueView = table.keyValueView();
+ List<BinaryRow> res = internalTable.getAll(keyRows, null).get();
- Publisher<BinaryRow> res = internalTable.scan(0, node.clock().now(),
node.node());
+ assertEquals(keyRows.size(), res.size());
- var subscriberAllDataAwaitLatch = new CountDownLatch(1);
+ var resIter = res.iterator();
- var retrievedItems = new ArrayList<BinaryRow>();
+ for (BinaryRowEx key : keyRows) {
+ int i =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor,
key)).<Long>value("key").intValue();
- res.subscribe(new Subscriber<>() {
- @Override
- public void onSubscribe(Subscription subscription) {
- subscription.request(10000);
- }
+ var resRow = resIter.next();
Review Comment:
It is incorrect to use `var` because the type is not visible from the line
of code.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItInternalTableTest.java:
##########
@@ -384,78 +359,190 @@ public void testRoScanAllExplicitPopulatingData() throws
InterruptedException {
roScanAll(false);
}
- private void roScanAll(boolean implicit) throws InterruptedException {
- IgniteImpl node = node();
+ @Test
+ public void getAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
Review Comment:
It is incorrect to use `var` because the type is not visible from the line
of code.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItInternalTableTest.java:
##########
@@ -384,78 +359,190 @@ public void testRoScanAllExplicitPopulatingData() throws
InterruptedException {
roScanAll(false);
}
- private void roScanAll(boolean implicit) throws InterruptedException {
- IgniteImpl node = node();
+ @Test
+ public void getAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
- KeyValueView<Tuple, Tuple> keyValueView = table.keyValueView();
+ List<BinaryRow> res = internalTable.getAll(keyRows, null).get();
Review Comment:
It is not correct to call `CompletableFuture#get` without a timeout, please
use
`org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher#willCompleteSuccessfully`.
##########
modules/client/src/test/java/org/apache/ignite/client/fakes/FakeInternalTable.java:
##########
@@ -348,7 +348,7 @@ public CompletableFuture<Collection<BinaryRow>>
deleteAll(Collection<BinaryRowEx
/** {@inheritDoc} */
Review Comment:
```suggestion
```
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItInternalTableTest.java:
##########
@@ -384,78 +359,190 @@ public void testRoScanAllExplicitPopulatingData() throws
InterruptedException {
roScanAll(false);
}
- private void roScanAll(boolean implicit) throws InterruptedException {
- IgniteImpl node = node();
+ @Test
+ public void getAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
Review Comment:
It is incorrect to use `var` because the type is not visible from the line
of code.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItInternalTableTest.java:
##########
@@ -384,78 +359,190 @@ public void testRoScanAllExplicitPopulatingData() throws
InterruptedException {
roScanAll(false);
}
- private void roScanAll(boolean implicit) throws InterruptedException {
- IgniteImpl node = node();
+ @Test
+ public void getAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
- KeyValueView<Tuple, Tuple> keyValueView = table.keyValueView();
+ List<BinaryRow> res = internalTable.getAll(keyRows, null).get();
- Publisher<BinaryRow> res = internalTable.scan(0, node.clock().now(),
node.node());
+ assertEquals(keyRows.size(), res.size());
- var subscriberAllDataAwaitLatch = new CountDownLatch(1);
+ var resIter = res.iterator();
- var retrievedItems = new ArrayList<BinaryRow>();
+ for (BinaryRowEx key : keyRows) {
+ int i =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor,
key)).<Long>value("key").intValue();
- res.subscribe(new Subscriber<>() {
- @Override
- public void onSubscribe(Subscription subscription) {
- subscription.request(10000);
- }
+ var resRow = resIter.next();
- @Override
- public void onNext(BinaryRow item) {
- retrievedItems.add(item);
- }
+ if (i % 2 == 1) {
+ assertNull(resRow);
+ } else {
+ assertNotNull(resRow);
- @Override
- public void onError(Throwable throwable) {
- fail("onError call is not expected.");
- }
+ var rowTuple =
TableRow.tuple(Row.wrapBinaryRow(schemaDescriptor, resRow));
Review Comment:
It is incorrect to use `var` because the type is not visible from the line
of code.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItInternalTableTest.java:
##########
@@ -384,78 +359,190 @@ public void testRoScanAllExplicitPopulatingData() throws
InterruptedException {
roScanAll(false);
}
- private void roScanAll(boolean implicit) throws InterruptedException {
- IgniteImpl node = node();
+ @Test
+ public void getAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
- KeyValueView<Tuple, Tuple> keyValueView = table.keyValueView();
+ List<BinaryRow> res = internalTable.getAll(keyRows, null).get();
- Publisher<BinaryRow> res = internalTable.scan(0, node.clock().now(),
node.node());
+ assertEquals(keyRows.size(), res.size());
- var subscriberAllDataAwaitLatch = new CountDownLatch(1);
+ var resIter = res.iterator();
- var retrievedItems = new ArrayList<BinaryRow>();
+ for (BinaryRowEx key : keyRows) {
+ int i =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor,
key)).<Long>value("key").intValue();
- res.subscribe(new Subscriber<>() {
- @Override
- public void onSubscribe(Subscription subscription) {
- subscription.request(10000);
- }
+ var resRow = resIter.next();
- @Override
- public void onNext(BinaryRow item) {
- retrievedItems.add(item);
- }
+ if (i % 2 == 1) {
+ assertNull(resRow);
+ } else {
+ assertNotNull(resRow);
- @Override
- public void onError(Throwable throwable) {
- fail("onError call is not expected.");
- }
+ var rowTuple =
TableRow.tuple(Row.wrapBinaryRow(schemaDescriptor, resRow));
- @Override
- public void onComplete() {
- subscriberAllDataAwaitLatch.countDown();
+ assertEquals(i % 100L, rowTuple.<Long>value("key"));
+ assertEquals(i, rowTuple.<Integer>value("valInt"));
+ assertEquals("some string row" + i,
rowTuple.<Integer>value("valStr"));
}
- });
+ }
+ }
- subscriberAllDataAwaitLatch.await();
+ @Test
+ public void deleteAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
- assertEquals(0, retrievedItems.size());
+ InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
- populateData(node, keyValueView, implicit);
+ List<BinaryRow> res = internalTable.deleteAll(keyRows, null).get();
+
+ var resIter = res.iterator();
Review Comment:
It is incorrect to use `var` because the type is not visible from the line
of code.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItInternalTableTest.java:
##########
@@ -384,78 +359,190 @@ public void testRoScanAllExplicitPopulatingData() throws
InterruptedException {
roScanAll(false);
}
- private void roScanAll(boolean implicit) throws InterruptedException {
- IgniteImpl node = node();
+ @Test
+ public void getAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
- KeyValueView<Tuple, Tuple> keyValueView = table.keyValueView();
+ List<BinaryRow> res = internalTable.getAll(keyRows, null).get();
- Publisher<BinaryRow> res = internalTable.scan(0, node.clock().now(),
node.node());
+ assertEquals(keyRows.size(), res.size());
- var subscriberAllDataAwaitLatch = new CountDownLatch(1);
+ var resIter = res.iterator();
- var retrievedItems = new ArrayList<BinaryRow>();
+ for (BinaryRowEx key : keyRows) {
+ int i =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor,
key)).<Long>value("key").intValue();
- res.subscribe(new Subscriber<>() {
- @Override
- public void onSubscribe(Subscription subscription) {
- subscription.request(10000);
- }
+ var resRow = resIter.next();
- @Override
- public void onNext(BinaryRow item) {
- retrievedItems.add(item);
- }
+ if (i % 2 == 1) {
+ assertNull(resRow);
+ } else {
+ assertNotNull(resRow);
- @Override
- public void onError(Throwable throwable) {
- fail("onError call is not expected.");
- }
+ var rowTuple =
TableRow.tuple(Row.wrapBinaryRow(schemaDescriptor, resRow));
- @Override
- public void onComplete() {
- subscriberAllDataAwaitLatch.countDown();
+ assertEquals(i % 100L, rowTuple.<Long>value("key"));
+ assertEquals(i, rowTuple.<Integer>value("valInt"));
+ assertEquals("some string row" + i,
rowTuple.<Integer>value("valStr"));
}
- });
+ }
+ }
- subscriberAllDataAwaitLatch.await();
+ @Test
+ public void deleteAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
- assertEquals(0, retrievedItems.size());
+ InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
Review Comment:
It is incorrect to use `var` because the type is not visible from the line
of code.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItInternalTableTest.java:
##########
@@ -384,78 +359,190 @@ public void testRoScanAllExplicitPopulatingData() throws
InterruptedException {
roScanAll(false);
}
- private void roScanAll(boolean implicit) throws InterruptedException {
- IgniteImpl node = node();
+ @Test
+ public void getAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
- KeyValueView<Tuple, Tuple> keyValueView = table.keyValueView();
+ List<BinaryRow> res = internalTable.getAll(keyRows, null).get();
- Publisher<BinaryRow> res = internalTable.scan(0, node.clock().now(),
node.node());
+ assertEquals(keyRows.size(), res.size());
- var subscriberAllDataAwaitLatch = new CountDownLatch(1);
+ var resIter = res.iterator();
- var retrievedItems = new ArrayList<BinaryRow>();
+ for (BinaryRowEx key : keyRows) {
+ int i =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor,
key)).<Long>value("key").intValue();
- res.subscribe(new Subscriber<>() {
- @Override
- public void onSubscribe(Subscription subscription) {
- subscription.request(10000);
- }
+ var resRow = resIter.next();
- @Override
- public void onNext(BinaryRow item) {
- retrievedItems.add(item);
- }
+ if (i % 2 == 1) {
+ assertNull(resRow);
+ } else {
+ assertNotNull(resRow);
- @Override
- public void onError(Throwable throwable) {
- fail("onError call is not expected.");
- }
+ var rowTuple =
TableRow.tuple(Row.wrapBinaryRow(schemaDescriptor, resRow));
- @Override
- public void onComplete() {
- subscriberAllDataAwaitLatch.countDown();
+ assertEquals(i % 100L, rowTuple.<Long>value("key"));
+ assertEquals(i, rowTuple.<Integer>value("valInt"));
+ assertEquals("some string row" + i,
rowTuple.<Integer>value("valStr"));
}
- });
+ }
+ }
- subscriberAllDataAwaitLatch.await();
+ @Test
+ public void deleteAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
- assertEquals(0, retrievedItems.size());
+ InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
- populateData(node, keyValueView, implicit);
+ List<BinaryRow> res = internalTable.deleteAll(keyRows, null).get();
+
+ var resIter = res.iterator();
- res = internalTable.scan(0, node.clock().now(), node.node());
+ for (BinaryRowEx key : keyRows) {
+ int i =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor,
key)).<Long>value("key").intValue();
- var subscriberAllDataAwaitLatch2 = new CountDownLatch(1);
+ if (i % 2 == 1) {
+ var rowTuple =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor, resIter.next()));
Review Comment:
It is incorrect to use `var` because the type is not visible from the line
of code.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItInternalTableTest.java:
##########
@@ -384,78 +359,190 @@ public void testRoScanAllExplicitPopulatingData() throws
InterruptedException {
roScanAll(false);
}
- private void roScanAll(boolean implicit) throws InterruptedException {
- IgniteImpl node = node();
+ @Test
+ public void getAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
- KeyValueView<Tuple, Tuple> keyValueView = table.keyValueView();
+ List<BinaryRow> res = internalTable.getAll(keyRows, null).get();
- Publisher<BinaryRow> res = internalTable.scan(0, node.clock().now(),
node.node());
+ assertEquals(keyRows.size(), res.size());
- var subscriberAllDataAwaitLatch = new CountDownLatch(1);
+ var resIter = res.iterator();
- var retrievedItems = new ArrayList<BinaryRow>();
+ for (BinaryRowEx key : keyRows) {
+ int i =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor,
key)).<Long>value("key").intValue();
- res.subscribe(new Subscriber<>() {
- @Override
- public void onSubscribe(Subscription subscription) {
- subscription.request(10000);
- }
+ var resRow = resIter.next();
- @Override
- public void onNext(BinaryRow item) {
- retrievedItems.add(item);
- }
+ if (i % 2 == 1) {
+ assertNull(resRow);
+ } else {
+ assertNotNull(resRow);
- @Override
- public void onError(Throwable throwable) {
- fail("onError call is not expected.");
- }
+ var rowTuple =
TableRow.tuple(Row.wrapBinaryRow(schemaDescriptor, resRow));
- @Override
- public void onComplete() {
- subscriberAllDataAwaitLatch.countDown();
+ assertEquals(i % 100L, rowTuple.<Long>value("key"));
+ assertEquals(i, rowTuple.<Integer>value("valInt"));
+ assertEquals("some string row" + i,
rowTuple.<Integer>value("valStr"));
}
- });
+ }
+ }
- subscriberAllDataAwaitLatch.await();
+ @Test
+ public void deleteAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
- assertEquals(0, retrievedItems.size());
+ InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
- populateData(node, keyValueView, implicit);
+ List<BinaryRow> res = internalTable.deleteAll(keyRows, null).get();
+
+ var resIter = res.iterator();
- res = internalTable.scan(0, node.clock().now(), node.node());
+ for (BinaryRowEx key : keyRows) {
+ int i =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor,
key)).<Long>value("key").intValue();
- var subscriberAllDataAwaitLatch2 = new CountDownLatch(1);
+ if (i % 2 == 1) {
+ var rowTuple =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor, resIter.next()));
- res.subscribe(new Subscriber<>() {
- @Override
- public void onSubscribe(Subscription subscription) {
- subscription.request(10000);
+ assertEquals(i % 100L, rowTuple.<Long>value("key"));
}
+ }
+ }
- @Override
- public void onNext(BinaryRow item) {
- retrievedItems.add(item);
+ @Test
+ public void deleteAllExactOrderTest() throws Exception {
+ var rowsToLookup = populateEvenKeysAndPrepareEntriesToLookup(false);
+
+ InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
Review Comment:
It is incorrect to use `var` because the type is not visible from the line
of code.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItInternalTableTest.java:
##########
@@ -384,78 +359,190 @@ public void testRoScanAllExplicitPopulatingData() throws
InterruptedException {
roScanAll(false);
}
- private void roScanAll(boolean implicit) throws InterruptedException {
- IgniteImpl node = node();
+ @Test
+ public void getAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
- KeyValueView<Tuple, Tuple> keyValueView = table.keyValueView();
+ List<BinaryRow> res = internalTable.getAll(keyRows, null).get();
- Publisher<BinaryRow> res = internalTable.scan(0, node.clock().now(),
node.node());
+ assertEquals(keyRows.size(), res.size());
- var subscriberAllDataAwaitLatch = new CountDownLatch(1);
+ var resIter = res.iterator();
- var retrievedItems = new ArrayList<BinaryRow>();
+ for (BinaryRowEx key : keyRows) {
+ int i =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor,
key)).<Long>value("key").intValue();
- res.subscribe(new Subscriber<>() {
- @Override
- public void onSubscribe(Subscription subscription) {
- subscription.request(10000);
- }
+ var resRow = resIter.next();
- @Override
- public void onNext(BinaryRow item) {
- retrievedItems.add(item);
- }
+ if (i % 2 == 1) {
+ assertNull(resRow);
+ } else {
+ assertNotNull(resRow);
- @Override
- public void onError(Throwable throwable) {
- fail("onError call is not expected.");
- }
+ var rowTuple =
TableRow.tuple(Row.wrapBinaryRow(schemaDescriptor, resRow));
- @Override
- public void onComplete() {
- subscriberAllDataAwaitLatch.countDown();
+ assertEquals(i % 100L, rowTuple.<Long>value("key"));
+ assertEquals(i, rowTuple.<Integer>value("valInt"));
+ assertEquals("some string row" + i,
rowTuple.<Integer>value("valStr"));
}
- });
+ }
+ }
- subscriberAllDataAwaitLatch.await();
+ @Test
+ public void deleteAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
Review Comment:
It is incorrect to use `var` because the type is not visible from the line
of code.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItInternalTableTest.java:
##########
@@ -384,78 +359,190 @@ public void testRoScanAllExplicitPopulatingData() throws
InterruptedException {
roScanAll(false);
}
- private void roScanAll(boolean implicit) throws InterruptedException {
- IgniteImpl node = node();
+ @Test
+ public void getAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
- KeyValueView<Tuple, Tuple> keyValueView = table.keyValueView();
+ List<BinaryRow> res = internalTable.getAll(keyRows, null).get();
- Publisher<BinaryRow> res = internalTable.scan(0, node.clock().now(),
node.node());
+ assertEquals(keyRows.size(), res.size());
- var subscriberAllDataAwaitLatch = new CountDownLatch(1);
+ var resIter = res.iterator();
- var retrievedItems = new ArrayList<BinaryRow>();
+ for (BinaryRowEx key : keyRows) {
+ int i =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor,
key)).<Long>value("key").intValue();
- res.subscribe(new Subscriber<>() {
- @Override
- public void onSubscribe(Subscription subscription) {
- subscription.request(10000);
- }
+ var resRow = resIter.next();
- @Override
- public void onNext(BinaryRow item) {
- retrievedItems.add(item);
- }
+ if (i % 2 == 1) {
+ assertNull(resRow);
+ } else {
+ assertNotNull(resRow);
- @Override
- public void onError(Throwable throwable) {
- fail("onError call is not expected.");
- }
+ var rowTuple =
TableRow.tuple(Row.wrapBinaryRow(schemaDescriptor, resRow));
- @Override
- public void onComplete() {
- subscriberAllDataAwaitLatch.countDown();
+ assertEquals(i % 100L, rowTuple.<Long>value("key"));
+ assertEquals(i, rowTuple.<Integer>value("valInt"));
+ assertEquals("some string row" + i,
rowTuple.<Integer>value("valStr"));
}
- });
+ }
+ }
- subscriberAllDataAwaitLatch.await();
+ @Test
+ public void deleteAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
- assertEquals(0, retrievedItems.size());
+ InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
- populateData(node, keyValueView, implicit);
+ List<BinaryRow> res = internalTable.deleteAll(keyRows, null).get();
+
+ var resIter = res.iterator();
- res = internalTable.scan(0, node.clock().now(), node.node());
+ for (BinaryRowEx key : keyRows) {
+ int i =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor,
key)).<Long>value("key").intValue();
- var subscriberAllDataAwaitLatch2 = new CountDownLatch(1);
+ if (i % 2 == 1) {
+ var rowTuple =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor, resIter.next()));
- res.subscribe(new Subscriber<>() {
- @Override
- public void onSubscribe(Subscription subscription) {
- subscription.request(10000);
+ assertEquals(i % 100L, rowTuple.<Long>value("key"));
}
+ }
+ }
- @Override
- public void onNext(BinaryRow item) {
- retrievedItems.add(item);
+ @Test
+ public void deleteAllExactOrderTest() throws Exception {
+ var rowsToLookup = populateEvenKeysAndPrepareEntriesToLookup(false);
+
+ InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
+
+ List<BinaryRow> res = internalTable.deleteAllExact(rowsToLookup,
null).get();
+
+ var resIter = res.iterator();
Review Comment:
It is incorrect to use `var` because the type is not visible from the line
of code.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItInternalTableTest.java:
##########
@@ -384,78 +359,190 @@ public void testRoScanAllExplicitPopulatingData() throws
InterruptedException {
roScanAll(false);
}
- private void roScanAll(boolean implicit) throws InterruptedException {
- IgniteImpl node = node();
+ @Test
+ public void getAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
- KeyValueView<Tuple, Tuple> keyValueView = table.keyValueView();
+ List<BinaryRow> res = internalTable.getAll(keyRows, null).get();
- Publisher<BinaryRow> res = internalTable.scan(0, node.clock().now(),
node.node());
+ assertEquals(keyRows.size(), res.size());
- var subscriberAllDataAwaitLatch = new CountDownLatch(1);
+ var resIter = res.iterator();
- var retrievedItems = new ArrayList<BinaryRow>();
+ for (BinaryRowEx key : keyRows) {
+ int i =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor,
key)).<Long>value("key").intValue();
- res.subscribe(new Subscriber<>() {
- @Override
- public void onSubscribe(Subscription subscription) {
- subscription.request(10000);
- }
+ var resRow = resIter.next();
- @Override
- public void onNext(BinaryRow item) {
- retrievedItems.add(item);
- }
+ if (i % 2 == 1) {
+ assertNull(resRow);
+ } else {
+ assertNotNull(resRow);
- @Override
- public void onError(Throwable throwable) {
- fail("onError call is not expected.");
- }
+ var rowTuple =
TableRow.tuple(Row.wrapBinaryRow(schemaDescriptor, resRow));
- @Override
- public void onComplete() {
- subscriberAllDataAwaitLatch.countDown();
+ assertEquals(i % 100L, rowTuple.<Long>value("key"));
+ assertEquals(i, rowTuple.<Integer>value("valInt"));
+ assertEquals("some string row" + i,
rowTuple.<Integer>value("valStr"));
}
- });
+ }
+ }
- subscriberAllDataAwaitLatch.await();
+ @Test
+ public void deleteAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
- assertEquals(0, retrievedItems.size());
+ InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
- populateData(node, keyValueView, implicit);
+ List<BinaryRow> res = internalTable.deleteAll(keyRows, null).get();
+
+ var resIter = res.iterator();
- res = internalTable.scan(0, node.clock().now(), node.node());
+ for (BinaryRowEx key : keyRows) {
+ int i =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor,
key)).<Long>value("key").intValue();
- var subscriberAllDataAwaitLatch2 = new CountDownLatch(1);
+ if (i % 2 == 1) {
+ var rowTuple =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor, resIter.next()));
- res.subscribe(new Subscriber<>() {
- @Override
- public void onSubscribe(Subscription subscription) {
- subscription.request(10000);
+ assertEquals(i % 100L, rowTuple.<Long>value("key"));
}
+ }
+ }
- @Override
- public void onNext(BinaryRow item) {
- retrievedItems.add(item);
+ @Test
+ public void deleteAllExactOrderTest() throws Exception {
+ var rowsToLookup = populateEvenKeysAndPrepareEntriesToLookup(false);
Review Comment:
It is incorrect to use `var` because the type is not visible from the line
of code.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItInternalTableTest.java:
##########
@@ -384,78 +359,190 @@ public void testRoScanAllExplicitPopulatingData() throws
InterruptedException {
roScanAll(false);
}
- private void roScanAll(boolean implicit) throws InterruptedException {
- IgniteImpl node = node();
+ @Test
+ public void getAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
- KeyValueView<Tuple, Tuple> keyValueView = table.keyValueView();
+ List<BinaryRow> res = internalTable.getAll(keyRows, null).get();
- Publisher<BinaryRow> res = internalTable.scan(0, node.clock().now(),
node.node());
+ assertEquals(keyRows.size(), res.size());
- var subscriberAllDataAwaitLatch = new CountDownLatch(1);
+ var resIter = res.iterator();
- var retrievedItems = new ArrayList<BinaryRow>();
+ for (BinaryRowEx key : keyRows) {
+ int i =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor,
key)).<Long>value("key").intValue();
- res.subscribe(new Subscriber<>() {
- @Override
- public void onSubscribe(Subscription subscription) {
- subscription.request(10000);
- }
+ var resRow = resIter.next();
- @Override
- public void onNext(BinaryRow item) {
- retrievedItems.add(item);
- }
+ if (i % 2 == 1) {
+ assertNull(resRow);
+ } else {
+ assertNotNull(resRow);
- @Override
- public void onError(Throwable throwable) {
- fail("onError call is not expected.");
- }
+ var rowTuple =
TableRow.tuple(Row.wrapBinaryRow(schemaDescriptor, resRow));
- @Override
- public void onComplete() {
- subscriberAllDataAwaitLatch.countDown();
+ assertEquals(i % 100L, rowTuple.<Long>value("key"));
+ assertEquals(i, rowTuple.<Integer>value("valInt"));
+ assertEquals("some string row" + i,
rowTuple.<Integer>value("valStr"));
}
- });
+ }
+ }
- subscriberAllDataAwaitLatch.await();
+ @Test
+ public void deleteAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
- assertEquals(0, retrievedItems.size());
+ InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
- populateData(node, keyValueView, implicit);
+ List<BinaryRow> res = internalTable.deleteAll(keyRows, null).get();
+
+ var resIter = res.iterator();
- res = internalTable.scan(0, node.clock().now(), node.node());
+ for (BinaryRowEx key : keyRows) {
+ int i =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor,
key)).<Long>value("key").intValue();
- var subscriberAllDataAwaitLatch2 = new CountDownLatch(1);
+ if (i % 2 == 1) {
+ var rowTuple =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor, resIter.next()));
- res.subscribe(new Subscriber<>() {
- @Override
- public void onSubscribe(Subscription subscription) {
- subscription.request(10000);
+ assertEquals(i % 100L, rowTuple.<Long>value("key"));
}
+ }
+ }
- @Override
- public void onNext(BinaryRow item) {
- retrievedItems.add(item);
+ @Test
+ public void deleteAllExactOrderTest() throws Exception {
+ var rowsToLookup = populateEvenKeysAndPrepareEntriesToLookup(false);
+
+ InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
+
+ List<BinaryRow> res = internalTable.deleteAllExact(rowsToLookup,
null).get();
Review Comment:
It is not correct to call `CompletableFuture#get` without a timeout, please
use
`org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher#willCompleteSuccessfully`.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItInternalTableTest.java:
##########
@@ -384,78 +359,190 @@ public void testRoScanAllExplicitPopulatingData() throws
InterruptedException {
roScanAll(false);
}
- private void roScanAll(boolean implicit) throws InterruptedException {
- IgniteImpl node = node();
+ @Test
+ public void getAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
- KeyValueView<Tuple, Tuple> keyValueView = table.keyValueView();
+ List<BinaryRow> res = internalTable.getAll(keyRows, null).get();
- Publisher<BinaryRow> res = internalTable.scan(0, node.clock().now(),
node.node());
+ assertEquals(keyRows.size(), res.size());
- var subscriberAllDataAwaitLatch = new CountDownLatch(1);
+ var resIter = res.iterator();
- var retrievedItems = new ArrayList<BinaryRow>();
+ for (BinaryRowEx key : keyRows) {
+ int i =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor,
key)).<Long>value("key").intValue();
- res.subscribe(new Subscriber<>() {
- @Override
- public void onSubscribe(Subscription subscription) {
- subscription.request(10000);
- }
+ var resRow = resIter.next();
- @Override
- public void onNext(BinaryRow item) {
- retrievedItems.add(item);
- }
+ if (i % 2 == 1) {
+ assertNull(resRow);
+ } else {
+ assertNotNull(resRow);
- @Override
- public void onError(Throwable throwable) {
- fail("onError call is not expected.");
- }
+ var rowTuple =
TableRow.tuple(Row.wrapBinaryRow(schemaDescriptor, resRow));
- @Override
- public void onComplete() {
- subscriberAllDataAwaitLatch.countDown();
+ assertEquals(i % 100L, rowTuple.<Long>value("key"));
+ assertEquals(i, rowTuple.<Integer>value("valInt"));
+ assertEquals("some string row" + i,
rowTuple.<Integer>value("valStr"));
}
- });
+ }
+ }
- subscriberAllDataAwaitLatch.await();
+ @Test
+ public void deleteAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
- assertEquals(0, retrievedItems.size());
+ InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
- populateData(node, keyValueView, implicit);
+ List<BinaryRow> res = internalTable.deleteAll(keyRows, null).get();
Review Comment:
It is not correct to call `CompletableFuture#get` without a timeout, please
use
`org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher#willCompleteSuccessfully`.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItInternalTableTest.java:
##########
@@ -384,78 +359,190 @@ public void testRoScanAllExplicitPopulatingData() throws
InterruptedException {
roScanAll(false);
}
- private void roScanAll(boolean implicit) throws InterruptedException {
- IgniteImpl node = node();
+ @Test
+ public void getAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
- KeyValueView<Tuple, Tuple> keyValueView = table.keyValueView();
+ List<BinaryRow> res = internalTable.getAll(keyRows, null).get();
- Publisher<BinaryRow> res = internalTable.scan(0, node.clock().now(),
node.node());
+ assertEquals(keyRows.size(), res.size());
- var subscriberAllDataAwaitLatch = new CountDownLatch(1);
+ var resIter = res.iterator();
- var retrievedItems = new ArrayList<BinaryRow>();
+ for (BinaryRowEx key : keyRows) {
+ int i =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor,
key)).<Long>value("key").intValue();
- res.subscribe(new Subscriber<>() {
- @Override
- public void onSubscribe(Subscription subscription) {
- subscription.request(10000);
- }
+ var resRow = resIter.next();
- @Override
- public void onNext(BinaryRow item) {
- retrievedItems.add(item);
- }
+ if (i % 2 == 1) {
+ assertNull(resRow);
+ } else {
+ assertNotNull(resRow);
- @Override
- public void onError(Throwable throwable) {
- fail("onError call is not expected.");
- }
+ var rowTuple =
TableRow.tuple(Row.wrapBinaryRow(schemaDescriptor, resRow));
- @Override
- public void onComplete() {
- subscriberAllDataAwaitLatch.countDown();
+ assertEquals(i % 100L, rowTuple.<Long>value("key"));
+ assertEquals(i, rowTuple.<Integer>value("valInt"));
+ assertEquals("some string row" + i,
rowTuple.<Integer>value("valStr"));
}
- });
+ }
+ }
- subscriberAllDataAwaitLatch.await();
+ @Test
+ public void deleteAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
- assertEquals(0, retrievedItems.size());
+ InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
- populateData(node, keyValueView, implicit);
+ List<BinaryRow> res = internalTable.deleteAll(keyRows, null).get();
+
+ var resIter = res.iterator();
- res = internalTable.scan(0, node.clock().now(), node.node());
+ for (BinaryRowEx key : keyRows) {
+ int i =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor,
key)).<Long>value("key").intValue();
- var subscriberAllDataAwaitLatch2 = new CountDownLatch(1);
+ if (i % 2 == 1) {
+ var rowTuple =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor, resIter.next()));
- res.subscribe(new Subscriber<>() {
- @Override
- public void onSubscribe(Subscription subscription) {
- subscription.request(10000);
+ assertEquals(i % 100L, rowTuple.<Long>value("key"));
}
+ }
+ }
- @Override
- public void onNext(BinaryRow item) {
- retrievedItems.add(item);
+ @Test
+ public void deleteAllExactOrderTest() throws Exception {
+ var rowsToLookup = populateEvenKeysAndPrepareEntriesToLookup(false);
+
+ InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
+
+ List<BinaryRow> res = internalTable.deleteAllExact(rowsToLookup,
null).get();
+
+ var resIter = res.iterator();
+
+ for (BinaryRowEx key : rowsToLookup) {
+ int i = TableRow.tuple(Row.wrapBinaryRow(schemaDescriptor,
key)).<Long>value("key").intValue();
+
+ if (i % 2 == 1) {
+ var rowTuple =
TableRow.tuple(Row.wrapBinaryRow(schemaDescriptor, resIter.next()));
+
+ assertEquals(i % 100L, rowTuple.<Long>value("key"));
+ assertEquals(i, rowTuple.<Integer>value("valInt"));
+ assertEquals("some string row" + i,
rowTuple.<Integer>value("valStr"));
}
+ }
+ }
+
+ @Test
+ public void insertAllOrderTest() throws Exception {
Review Comment:
Your task is only about deleting, not about inserting data.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItInternalTableTest.java:
##########
@@ -384,78 +359,190 @@ public void testRoScanAllExplicitPopulatingData() throws
InterruptedException {
roScanAll(false);
}
- private void roScanAll(boolean implicit) throws InterruptedException {
- IgniteImpl node = node();
+ @Test
+ public void getAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
- KeyValueView<Tuple, Tuple> keyValueView = table.keyValueView();
+ List<BinaryRow> res = internalTable.getAll(keyRows, null).get();
- Publisher<BinaryRow> res = internalTable.scan(0, node.clock().now(),
node.node());
+ assertEquals(keyRows.size(), res.size());
- var subscriberAllDataAwaitLatch = new CountDownLatch(1);
+ var resIter = res.iterator();
- var retrievedItems = new ArrayList<BinaryRow>();
+ for (BinaryRowEx key : keyRows) {
+ int i =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor,
key)).<Long>value("key").intValue();
- res.subscribe(new Subscriber<>() {
- @Override
- public void onSubscribe(Subscription subscription) {
- subscription.request(10000);
- }
+ var resRow = resIter.next();
- @Override
- public void onNext(BinaryRow item) {
- retrievedItems.add(item);
- }
+ if (i % 2 == 1) {
+ assertNull(resRow);
+ } else {
+ assertNotNull(resRow);
- @Override
- public void onError(Throwable throwable) {
- fail("onError call is not expected.");
- }
+ var rowTuple =
TableRow.tuple(Row.wrapBinaryRow(schemaDescriptor, resRow));
- @Override
- public void onComplete() {
- subscriberAllDataAwaitLatch.countDown();
+ assertEquals(i % 100L, rowTuple.<Long>value("key"));
+ assertEquals(i, rowTuple.<Integer>value("valInt"));
+ assertEquals("some string row" + i,
rowTuple.<Integer>value("valStr"));
}
- });
+ }
+ }
- subscriberAllDataAwaitLatch.await();
+ @Test
+ public void deleteAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
- assertEquals(0, retrievedItems.size());
+ InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
- populateData(node, keyValueView, implicit);
+ List<BinaryRow> res = internalTable.deleteAll(keyRows, null).get();
+
+ var resIter = res.iterator();
- res = internalTable.scan(0, node.clock().now(), node.node());
+ for (BinaryRowEx key : keyRows) {
+ int i =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor,
key)).<Long>value("key").intValue();
- var subscriberAllDataAwaitLatch2 = new CountDownLatch(1);
+ if (i % 2 == 1) {
+ var rowTuple =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor, resIter.next()));
- res.subscribe(new Subscriber<>() {
- @Override
- public void onSubscribe(Subscription subscription) {
- subscription.request(10000);
+ assertEquals(i % 100L, rowTuple.<Long>value("key"));
}
+ }
+ }
- @Override
- public void onNext(BinaryRow item) {
- retrievedItems.add(item);
+ @Test
+ public void deleteAllExactOrderTest() throws Exception {
+ var rowsToLookup = populateEvenKeysAndPrepareEntriesToLookup(false);
+
+ InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
+
+ List<BinaryRow> res = internalTable.deleteAllExact(rowsToLookup,
null).get();
+
+ var resIter = res.iterator();
+
+ for (BinaryRowEx key : rowsToLookup) {
+ int i = TableRow.tuple(Row.wrapBinaryRow(schemaDescriptor,
key)).<Long>value("key").intValue();
+
+ if (i % 2 == 1) {
+ var rowTuple =
TableRow.tuple(Row.wrapBinaryRow(schemaDescriptor, resIter.next()));
Review Comment:
It is incorrect to use `var` because the type is not visible from the line
of code.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/InternalTable.java:
##########
@@ -181,7 +181,7 @@ CompletableFuture<List<BinaryRow>> getAll(
* @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- CompletableFuture<Collection<BinaryRow>> insertAll(Collection<BinaryRowEx>
rows, @Nullable InternalTransaction tx);
+ CompletableFuture<List<BinaryRow>> insertAll(Collection<BinaryRowEx> rows,
@Nullable InternalTransaction tx);
Review Comment:
Your task is only about deleting, not about inserting data.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/InternalTable.java:
##########
@@ -246,7 +246,7 @@ CompletableFuture<List<BinaryRow>> getAll(
* @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- CompletableFuture<Collection<BinaryRow>> deleteAll(Collection<BinaryRowEx>
rows, @Nullable InternalTransaction tx);
+ CompletableFuture<List<BinaryRow>> deleteAll(Collection<BinaryRowEx> rows,
@Nullable InternalTransaction tx);
Review Comment:
It is necessary to change the documentation and describe what will be
returned in the future with examples.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java:
##########
@@ -948,7 +948,7 @@ public CompletableFuture<Boolean> insert(BinaryRowEx row,
InternalTransaction tx
/** {@inheritDoc} */
@Override
- public CompletableFuture<Collection<BinaryRow>>
insertAll(Collection<BinaryRowEx> rows, InternalTransaction tx) {
+ public CompletableFuture<List<BinaryRow>>
insertAll(Collection<BinaryRowEx> rows, InternalTransaction tx) {
Review Comment:
Your task is only about deleting, not about inserting data.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItInternalTableTest.java:
##########
@@ -384,78 +359,190 @@ public void testRoScanAllExplicitPopulatingData() throws
InterruptedException {
roScanAll(false);
}
- private void roScanAll(boolean implicit) throws InterruptedException {
- IgniteImpl node = node();
+ @Test
+ public void getAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
- KeyValueView<Tuple, Tuple> keyValueView = table.keyValueView();
+ List<BinaryRow> res = internalTable.getAll(keyRows, null).get();
- Publisher<BinaryRow> res = internalTable.scan(0, node.clock().now(),
node.node());
+ assertEquals(keyRows.size(), res.size());
- var subscriberAllDataAwaitLatch = new CountDownLatch(1);
+ var resIter = res.iterator();
- var retrievedItems = new ArrayList<BinaryRow>();
+ for (BinaryRowEx key : keyRows) {
+ int i =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor,
key)).<Long>value("key").intValue();
- res.subscribe(new Subscriber<>() {
- @Override
- public void onSubscribe(Subscription subscription) {
- subscription.request(10000);
- }
+ var resRow = resIter.next();
- @Override
- public void onNext(BinaryRow item) {
- retrievedItems.add(item);
- }
+ if (i % 2 == 1) {
+ assertNull(resRow);
+ } else {
+ assertNotNull(resRow);
- @Override
- public void onError(Throwable throwable) {
- fail("onError call is not expected.");
- }
+ var rowTuple =
TableRow.tuple(Row.wrapBinaryRow(schemaDescriptor, resRow));
- @Override
- public void onComplete() {
- subscriberAllDataAwaitLatch.countDown();
+ assertEquals(i % 100L, rowTuple.<Long>value("key"));
+ assertEquals(i, rowTuple.<Integer>value("valInt"));
+ assertEquals("some string row" + i,
rowTuple.<Integer>value("valStr"));
}
- });
+ }
+ }
- subscriberAllDataAwaitLatch.await();
+ @Test
+ public void deleteAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
- assertEquals(0, retrievedItems.size());
+ InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
- populateData(node, keyValueView, implicit);
+ List<BinaryRow> res = internalTable.deleteAll(keyRows, null).get();
+
+ var resIter = res.iterator();
- res = internalTable.scan(0, node.clock().now(), node.node());
+ for (BinaryRowEx key : keyRows) {
+ int i =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor,
key)).<Long>value("key").intValue();
- var subscriberAllDataAwaitLatch2 = new CountDownLatch(1);
+ if (i % 2 == 1) {
+ var rowTuple =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor, resIter.next()));
- res.subscribe(new Subscriber<>() {
- @Override
- public void onSubscribe(Subscription subscription) {
- subscription.request(10000);
+ assertEquals(i % 100L, rowTuple.<Long>value("key"));
}
+ }
+ }
- @Override
- public void onNext(BinaryRow item) {
- retrievedItems.add(item);
+ @Test
+ public void deleteAllExactOrderTest() throws Exception {
+ var rowsToLookup = populateEvenKeysAndPrepareEntriesToLookup(false);
+
+ InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
+
+ List<BinaryRow> res = internalTable.deleteAllExact(rowsToLookup,
null).get();
+
+ var resIter = res.iterator();
+
+ for (BinaryRowEx key : rowsToLookup) {
+ int i = TableRow.tuple(Row.wrapBinaryRow(schemaDescriptor,
key)).<Long>value("key").intValue();
+
+ if (i % 2 == 1) {
+ var rowTuple =
TableRow.tuple(Row.wrapBinaryRow(schemaDescriptor, resIter.next()));
+
+ assertEquals(i % 100L, rowTuple.<Long>value("key"));
+ assertEquals(i, rowTuple.<Integer>value("valInt"));
+ assertEquals("some string row" + i,
rowTuple.<Integer>value("valStr"));
}
+ }
+ }
+
+ @Test
+ public void insertAllOrderTest() throws Exception {
+ var rowsToLookup = populateEvenKeysAndPrepareEntriesToLookup(false);
+
+ InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
+
+ List<BinaryRow> res = internalTable.insertAll(rowsToLookup,
null).get();
+
+ var resIter = res.iterator();
- @Override
- public void onError(Throwable throwable) {
- fail("onError call is not expected.");
+ for (BinaryRowEx key : rowsToLookup) {
+ int i = TableRow.tuple(Row.wrapBinaryRow(schemaDescriptor,
key)).<Long>value("key").intValue();
+
+ if (i % 2 == 0) {
+ var rowTuple =
TableRow.tuple(Row.wrapBinaryRow(schemaDescriptor, resIter.next()));
+
+ assertEquals(i % 100L, rowTuple.<Long>value("key"));
+ assertEquals(i, rowTuple.<Integer>value("valInt"));
+ assertEquals("some string row" + i,
rowTuple.<Integer>value("valStr"));
}
+ }
+ }
+
+ private ArrayList<BinaryRowEx>
populateEvenKeysAndPrepareEntriesToLookup(boolean keyOnly) {
+ KeyValueView<Tuple, Tuple> keyValueView = table.keyValueView();
+
+ var keyRows = new ArrayList<BinaryRowEx>();
- @Override
- public void onComplete() {
- subscriberAllDataAwaitLatch2.countDown();
+ for (int i = 0; i < 15; i++) {
+ keyRows.add(keyOnly ? createKeyRow(i) : createKeyValueRow(i, i,
"some string row" + i));
+
+ if (i % 2 == 0) {
+ putValue(keyValueView, i);
}
- });
+ }
+
+ Collections.shuffle(keyRows);
- subscriberAllDataAwaitLatch2.await();
+ return keyRows;
+ }
+
+ private void roScanAll(boolean implicit) throws InterruptedException {
+ IgniteImpl node = node();
+
+ KeyValueView<Tuple, Tuple> keyValueView = table.keyValueView();
+
+ var retrievedItems = scanAllPartitions(node);
Review Comment:
It is incorrect to use `var` because the type is not visible from the line
of code.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItInternalTableTest.java:
##########
@@ -384,78 +359,190 @@ public void testRoScanAllExplicitPopulatingData() throws
InterruptedException {
roScanAll(false);
}
- private void roScanAll(boolean implicit) throws InterruptedException {
- IgniteImpl node = node();
+ @Test
+ public void getAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
- KeyValueView<Tuple, Tuple> keyValueView = table.keyValueView();
+ List<BinaryRow> res = internalTable.getAll(keyRows, null).get();
- Publisher<BinaryRow> res = internalTable.scan(0, node.clock().now(),
node.node());
+ assertEquals(keyRows.size(), res.size());
- var subscriberAllDataAwaitLatch = new CountDownLatch(1);
+ var resIter = res.iterator();
- var retrievedItems = new ArrayList<BinaryRow>();
+ for (BinaryRowEx key : keyRows) {
+ int i =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor,
key)).<Long>value("key").intValue();
- res.subscribe(new Subscriber<>() {
- @Override
- public void onSubscribe(Subscription subscription) {
- subscription.request(10000);
- }
+ var resRow = resIter.next();
- @Override
- public void onNext(BinaryRow item) {
- retrievedItems.add(item);
- }
+ if (i % 2 == 1) {
+ assertNull(resRow);
+ } else {
+ assertNotNull(resRow);
- @Override
- public void onError(Throwable throwable) {
- fail("onError call is not expected.");
- }
+ var rowTuple =
TableRow.tuple(Row.wrapBinaryRow(schemaDescriptor, resRow));
- @Override
- public void onComplete() {
- subscriberAllDataAwaitLatch.countDown();
+ assertEquals(i % 100L, rowTuple.<Long>value("key"));
+ assertEquals(i, rowTuple.<Integer>value("valInt"));
+ assertEquals("some string row" + i,
rowTuple.<Integer>value("valStr"));
}
- });
+ }
+ }
- subscriberAllDataAwaitLatch.await();
+ @Test
+ public void deleteAllOrderTest() throws Exception {
+ var keyRows = populateEvenKeysAndPrepareEntriesToLookup(true);
- assertEquals(0, retrievedItems.size());
+ InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
- populateData(node, keyValueView, implicit);
+ List<BinaryRow> res = internalTable.deleteAll(keyRows, null).get();
+
+ var resIter = res.iterator();
- res = internalTable.scan(0, node.clock().now(), node.node());
+ for (BinaryRowEx key : keyRows) {
+ int i =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor,
key)).<Long>value("key").intValue();
- var subscriberAllDataAwaitLatch2 = new CountDownLatch(1);
+ if (i % 2 == 1) {
+ var rowTuple =
TableRow.keyTuple(Row.wrapKeyOnlyBinaryRow(schemaDescriptor, resIter.next()));
- res.subscribe(new Subscriber<>() {
- @Override
- public void onSubscribe(Subscription subscription) {
- subscription.request(10000);
+ assertEquals(i % 100L, rowTuple.<Long>value("key"));
}
+ }
+ }
- @Override
- public void onNext(BinaryRow item) {
- retrievedItems.add(item);
+ @Test
+ public void deleteAllExactOrderTest() throws Exception {
+ var rowsToLookup = populateEvenKeysAndPrepareEntriesToLookup(false);
+
+ InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
+
+ List<BinaryRow> res = internalTable.deleteAllExact(rowsToLookup,
null).get();
+
+ var resIter = res.iterator();
+
+ for (BinaryRowEx key : rowsToLookup) {
+ int i = TableRow.tuple(Row.wrapBinaryRow(schemaDescriptor,
key)).<Long>value("key").intValue();
+
+ if (i % 2 == 1) {
+ var rowTuple =
TableRow.tuple(Row.wrapBinaryRow(schemaDescriptor, resIter.next()));
+
+ assertEquals(i % 100L, rowTuple.<Long>value("key"));
+ assertEquals(i, rowTuple.<Integer>value("valInt"));
+ assertEquals("some string row" + i,
rowTuple.<Integer>value("valStr"));
}
+ }
+ }
+
+ @Test
+ public void insertAllOrderTest() throws Exception {
+ var rowsToLookup = populateEvenKeysAndPrepareEntriesToLookup(false);
+
+ InternalTable internalTable = ((TableImpl) table).internalTable();
+ var schemaDescriptor = ((TableImpl) table).schemaView().schema();
+
+ List<BinaryRow> res = internalTable.insertAll(rowsToLookup,
null).get();
+
+ var resIter = res.iterator();
- @Override
- public void onError(Throwable throwable) {
- fail("onError call is not expected.");
+ for (BinaryRowEx key : rowsToLookup) {
+ int i = TableRow.tuple(Row.wrapBinaryRow(schemaDescriptor,
key)).<Long>value("key").intValue();
+
+ if (i % 2 == 0) {
+ var rowTuple =
TableRow.tuple(Row.wrapBinaryRow(schemaDescriptor, resIter.next()));
+
+ assertEquals(i % 100L, rowTuple.<Long>value("key"));
+ assertEquals(i, rowTuple.<Integer>value("valInt"));
+ assertEquals("some string row" + i,
rowTuple.<Integer>value("valStr"));
}
+ }
+ }
+
+ private ArrayList<BinaryRowEx>
populateEvenKeysAndPrepareEntriesToLookup(boolean keyOnly) {
+ KeyValueView<Tuple, Tuple> keyValueView = table.keyValueView();
+
+ var keyRows = new ArrayList<BinaryRowEx>();
- @Override
- public void onComplete() {
- subscriberAllDataAwaitLatch2.countDown();
+ for (int i = 0; i < 15; i++) {
+ keyRows.add(keyOnly ? createKeyRow(i) : createKeyValueRow(i, i,
"some string row" + i));
+
+ if (i % 2 == 0) {
+ putValue(keyValueView, i);
}
- });
+ }
+
+ Collections.shuffle(keyRows);
- subscriberAllDataAwaitLatch2.await();
+ return keyRows;
+ }
+
+ private void roScanAll(boolean implicit) throws InterruptedException {
+ IgniteImpl node = node();
+
+ KeyValueView<Tuple, Tuple> keyValueView = table.keyValueView();
+
+ var retrievedItems = scanAllPartitions(node);
+
+ assertEquals(0, retrievedItems.size());
+
+ populateData(node, keyValueView, implicit);
+
+ retrievedItems = scanAllPartitions(node);
assertEquals(15, retrievedItems.size());
}
+
+ /**
+ * Scans all table entries.
+ *
+ * @param node Ignite instance.
+ * @return Collection with all rows.
+ * @throws InterruptedException If fail.
+ */
+ private static List<BinaryRow> scanAllPartitions(IgniteImpl node) throws
InterruptedException {
+ InternalTable internalTable = ((TableImpl)
node.tables().table(TABLE_NAME)).internalTable();
+
+ List<BinaryRow> retrievedItems = new CopyOnWriteArrayList<>();
+
+ int parts = internalTable.partitions();
+
+ var subscriberAllDataAwaitLatch = new CountDownLatch(parts);
+
+ for (int i = 0; i < parts; i++) {
+ Publisher<BinaryRow> res = internalTable.scan(i,
node.clock().now(), node.node());
+
+ res.subscribe(new Subscriber<>() {
+ @Override
+ public void onSubscribe(Subscription subscription) {
+ subscription.request(10000);
+ }
+
+ @Override
+ public void onNext(BinaryRow item) {
+ retrievedItems.add(item);
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ fail("onError call is not expected.");
+ }
+
+ @Override
+ public void onComplete() {
+ subscriberAllDataAwaitLatch.countDown();
+ }
+ });
+ }
+
+ subscriberAllDataAwaitLatch.await();
Review Comment:
Waiting without a timeout is dangerous.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java:
##########
@@ -1485,44 +1468,64 @@ 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();
- if (batchResult == null) {
- continue;
- }
+ assert batch.requestedRows.size() == response.size();
Review Comment:
Please indicate their sizes in the error message.
##########
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:
Why public?
##########
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:
Please don't use the stream.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/InternalTable.java:
##########
@@ -255,7 +255,7 @@ CompletableFuture<List<BinaryRow>> getAll(
* @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- CompletableFuture<Collection<BinaryRow>>
deleteAllExact(Collection<BinaryRowEx> rows, @Nullable InternalTransaction tx);
+ CompletableFuture<List<BinaryRow>> deleteAllExact(Collection<BinaryRowEx>
rows, @Nullable InternalTransaction tx);
Review Comment:
It is necessary to change the documentation and describe what will be
returned in the future with examples.
##########
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:
U can use for(int i=0; ...
##########
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:
Please don't use stream.
--
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
