You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by as...@apache.org on 2021/07/15 14:14:12 UTC
[ignite-3] branch main updated: IGNITE-15086 Public transactions
API - Fixes #214.
This is an automated email from the ASF dual-hosted git repository.
ascherbakov pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new b67df02 IGNITE-15086 Public transactions API - Fixes #214.
b67df02 is described below
commit b67df02af17bcd56c31e0b04fc167344cfeb313e
Author: Alexey Scherbakov <al...@gmail.com>
AuthorDate: Thu Jul 15 17:13:38 2021 +0300
IGNITE-15086 Public transactions API - Fixes #214.
Signed-off-by: Alexey Scherbakov <al...@gmail.com>
---
.../main/java/org/apache/ignite/app/Ignite.java | 8 +
.../apache/ignite/table/KeyValueBinaryView.java | 5 +
.../java/org/apache/ignite/table/KeyValueView.java | 15 ++
.../java/org/apache/ignite/table/RecordView.java | 4 +
.../main/java/org/apache/ignite/table/Table.java | 12 ++
.../java/org/apache/ignite/table/TableView.java | 15 ++
.../Ignite.java => tx/IgniteTransactions.java} | 24 ++-
.../java/org/apache/ignite/tx/Transaction.java | 53 ++++++
.../raft/jraft/conf/ConfigurationManagerTest.java | 1 -
.../ignite/raft/jraft/conf/ConfigurationTest.java | 1 -
.../ignite/raft/jraft/test/MockAsyncContext.java | 2 +-
.../org/apache/ignite/internal/app/IgniteImpl.java | 6 +
.../ignite/distributed/ITDistributedTableTest.java | 3 +-
.../ignite/internal/table/AbstractTableView.java | 22 ++-
.../ignite/internal/table/InternalTable.java | 48 +++--
.../ignite/internal/table/KVBinaryViewImpl.java | 42 +++--
.../apache/ignite/internal/table/KVViewImpl.java | 15 +-
.../ignite/internal/table/RecordViewImpl.java | 19 +-
.../apache/ignite/internal/table/TableImpl.java | 49 +++--
.../internal/table/distributed/TableManager.java | 2 +-
.../distributed/storage/InternalTableImpl.java | 35 ++--
.../org/apache/ignite/internal/table/Example.java | 2 +-
.../internal/table/KVViewOperationsTest.java | 21 ++-
.../internal/table/StrictSchemaOperationsTest.java | 8 +-
.../table/TableBinaryViewOperationsTest.java | 18 +-
.../org/apache/ignite/internal/table/TxTest.java | 208 +++++++++++++++++++++
.../table/impl/DummyInternalTableImpl.java | 34 ++--
27 files changed, 544 insertions(+), 128 deletions(-)
diff --git a/modules/api/src/main/java/org/apache/ignite/app/Ignite.java b/modules/api/src/main/java/org/apache/ignite/app/Ignite.java
index 2c98cfd..87e585a 100644
--- a/modules/api/src/main/java/org/apache/ignite/app/Ignite.java
+++ b/modules/api/src/main/java/org/apache/ignite/app/Ignite.java
@@ -18,6 +18,7 @@
package org.apache.ignite.app;
import org.apache.ignite.table.manager.IgniteTables;
+import org.apache.ignite.tx.IgniteTransactions;
/**
* Ignite node interface. Main entry-point for all Ignite APIs.
@@ -29,4 +30,11 @@ public interface Ignite extends AutoCloseable {
* @return Ignite tables.
*/
IgniteTables tables();
+
+ /**
+ * Returns a transaction facade.
+ *
+ * @return Ignite transactions.
+ */
+ IgniteTransactions transactions();
}
diff --git a/modules/api/src/main/java/org/apache/ignite/table/KeyValueBinaryView.java b/modules/api/src/main/java/org/apache/ignite/table/KeyValueBinaryView.java
index 75b95db..fcf04af 100644
--- a/modules/api/src/main/java/org/apache/ignite/table/KeyValueBinaryView.java
+++ b/modules/api/src/main/java/org/apache/ignite/table/KeyValueBinaryView.java
@@ -17,6 +17,8 @@
package org.apache.ignite.table;
+import org.apache.ignite.tx.Transaction;
+
/**
* Key-value view of table provides methods to access the data using key-value approach and
* regarding the binary object concept.
@@ -28,4 +30,7 @@ public interface KeyValueBinaryView extends KeyValueView<Tuple, Tuple> {
* @return Tuple builder.
*/
TupleBuilder tupleBuilder();
+
+ /** {@inheritDoc} */
+ @Override KeyValueBinaryView withTransaction(Transaction tx);
}
diff --git a/modules/api/src/main/java/org/apache/ignite/table/KeyValueView.java b/modules/api/src/main/java/org/apache/ignite/table/KeyValueView.java
index ef0a80d..f7cc98c 100644
--- a/modules/api/src/main/java/org/apache/ignite/table/KeyValueView.java
+++ b/modules/api/src/main/java/org/apache/ignite/table/KeyValueView.java
@@ -22,7 +22,9 @@ import java.util.Collection;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.table.mapper.Mappers;
+import org.apache.ignite.tx.Transaction;
import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
/**
* Key-Value view of table provides methods to access table records.
@@ -379,4 +381,17 @@ public interface KeyValueView<K, V> {
@NotNull Collection<K> keys,
InvokeProcessor<K, V, R> proc,
Serializable... args);
+
+ /**
+ * @return Current transaction or null if the view is not enlisted in a transaction.
+ */
+ @Nullable Transaction transaction();
+
+ /**
+ * Enslists a view into the transaction.
+ *
+ * @param tx The transaction.
+ * @return Transactional view.
+ */
+ KeyValueView<K, V> withTransaction(Transaction tx);
}
diff --git a/modules/api/src/main/java/org/apache/ignite/table/RecordView.java b/modules/api/src/main/java/org/apache/ignite/table/RecordView.java
index 102da62..eb404be 100644
--- a/modules/api/src/main/java/org/apache/ignite/table/RecordView.java
+++ b/modules/api/src/main/java/org/apache/ignite/table/RecordView.java
@@ -19,6 +19,7 @@ package org.apache.ignite.table;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.table.mapper.Mappers;
+import org.apache.ignite.tx.Transaction;
/**
* Record view of table provides methods to access table records.
@@ -52,4 +53,7 @@ public interface RecordView<R> extends TableView<R> {
* @return Future representing pending completion of the operation.
*/
CompletableFuture<R> fillAsync(R recObjToFill);
+
+ /** {@inheritDoc} */
+ @Override RecordView<R> withTransaction(Transaction tx);
}
diff --git a/modules/api/src/main/java/org/apache/ignite/table/Table.java b/modules/api/src/main/java/org/apache/ignite/table/Table.java
index 8c9cf5c..5a8974b 100644
--- a/modules/api/src/main/java/org/apache/ignite/table/Table.java
+++ b/modules/api/src/main/java/org/apache/ignite/table/Table.java
@@ -21,6 +21,8 @@ import org.apache.ignite.table.mapper.KeyMapper;
import org.apache.ignite.table.mapper.Mappers;
import org.apache.ignite.table.mapper.RecordMapper;
import org.apache.ignite.table.mapper.ValueMapper;
+import org.apache.ignite.tx.IgniteTransactions;
+import org.apache.ignite.tx.Transaction;
import org.jetbrains.annotations.NotNull;
/**
@@ -74,6 +76,16 @@ public interface Table extends TableView<Tuple> {
KeyValueBinaryView kvView();
/**
+ * Creates a transactional view of the table.
+ *
+ * @param tx The transaction.
+ * @return Transactional table view.
+ * @see Transaction
+ * @see IgniteTransactions
+ */
+ @Override Table withTransaction(Transaction tx);
+
+ /**
* Creates record view of table for record class provided.
*
* @param recCls Record class.
diff --git a/modules/api/src/main/java/org/apache/ignite/table/TableView.java b/modules/api/src/main/java/org/apache/ignite/table/TableView.java
index 6ea9109..77258e8 100644
--- a/modules/api/src/main/java/org/apache/ignite/table/TableView.java
+++ b/modules/api/src/main/java/org/apache/ignite/table/TableView.java
@@ -21,7 +21,9 @@ import java.io.Serializable;
import java.util.Collection;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.tx.Transaction;
import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
/**
* Table view interface provides methods to access table records.
@@ -351,4 +353,17 @@ public interface TableView<R> {
*/
@NotNull <T extends Serializable> CompletableFuture<Map<R, T>> invokeAllAsync(@NotNull Collection<R> keyRecs,
InvokeProcessor<R, R, T> proc);
+
+ /**
+ * @return Current transaction or null if a table is not enlisted in a transaction.
+ */
+ @Nullable Transaction transaction();
+
+ /**
+ * Enslists a view into the transaction.
+ *
+ * @param tx The transaction.
+ * @return Transactional view.
+ */
+ TableView<R> withTransaction(Transaction tx);
}
diff --git a/modules/api/src/main/java/org/apache/ignite/app/Ignite.java b/modules/api/src/main/java/org/apache/ignite/tx/IgniteTransactions.java
similarity index 59%
copy from modules/api/src/main/java/org/apache/ignite/app/Ignite.java
copy to modules/api/src/main/java/org/apache/ignite/tx/IgniteTransactions.java
index 2c98cfd..e659133 100644
--- a/modules/api/src/main/java/org/apache/ignite/app/Ignite.java
+++ b/modules/api/src/main/java/org/apache/ignite/tx/IgniteTransactions.java
@@ -15,18 +15,28 @@
* limitations under the License.
*/
-package org.apache.ignite.app;
+package org.apache.ignite.tx;
-import org.apache.ignite.table.manager.IgniteTables;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
/**
- * Ignite node interface. Main entry-point for all Ignite APIs.
+ * Ignite Transactions facade.
*/
-public interface Ignite extends AutoCloseable {
+public interface IgniteTransactions {
/**
- * Gets an object for manipulate Ignite tables.
+ * Begins a transaction.
*
- * @return Ignite tables.
+ * @return The future.
*/
- IgniteTables tables();
+ CompletableFuture<Transaction> beginAsync();
+
+ /**
+ * Synchronously executes a closure within a transaction.
+ * <p>
+ * If the closure is executed normally (no exceptions), the transaction is automatically committed.
+ *
+ * @param clo The closure.
+ */
+ void runInTransaction(Consumer<Transaction> clo);
}
diff --git a/modules/api/src/main/java/org/apache/ignite/tx/Transaction.java b/modules/api/src/main/java/org/apache/ignite/tx/Transaction.java
new file mode 100644
index 0000000..30a2e28
--- /dev/null
+++ b/modules/api/src/main/java/org/apache/ignite/tx/Transaction.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.tx;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * The transaction.
+ */
+public interface Transaction {
+ /**
+ * Synchronously commits a transaction.
+ * Does nothing if it's already finished by commiting or rolling back.
+ */
+ void commit();
+
+ /**
+ * Asynchronously commits a transaction.
+ * Does nothing if it's already finished by commiting or rolling back.
+ *
+ * @return The future.
+ */
+ CompletableFuture<Void> commitAsync();
+
+ /**
+ * Synchronously rolls back a transaction.
+ * Does nothing if it's already finished by commiting or rolling back.
+ */
+ void rollback();
+
+ /**
+ * Asynchronously rolls back a transaction.
+ * Does nothing if it's already finished by commiting or rolling back.
+ *
+ * @return The future.
+ */
+ CompletableFuture<Void> rollbackAsync();
+}
diff --git a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/conf/ConfigurationManagerTest.java b/modules/raft/src/test/java/org/apache/ignite/raft/jraft/conf/ConfigurationManagerTest.java
index a14fda5..af96b74 100644
--- a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/conf/ConfigurationManagerTest.java
+++ b/modules/raft/src/test/java/org/apache/ignite/raft/jraft/conf/ConfigurationManagerTest.java
@@ -27,7 +27,6 @@ import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class ConfigurationManagerTest {
-
private ConfigurationManager confManager;
@BeforeEach
diff --git a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/conf/ConfigurationTest.java b/modules/raft/src/test/java/org/apache/ignite/raft/jraft/conf/ConfigurationTest.java
index e2ca3fd..ef7bf6a 100644
--- a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/conf/ConfigurationTest.java
+++ b/modules/raft/src/test/java/org/apache/ignite/raft/jraft/conf/ConfigurationTest.java
@@ -26,7 +26,6 @@ import static org.junit.jupiter.api.Assertions.assertNotSame;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class ConfigurationTest {
-
@Test
public void testToStringParseStuff() {
final String confStr = "localhost:8081,localhost:8082,localhost:8083";
diff --git a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/test/MockAsyncContext.java b/modules/raft/src/test/java/org/apache/ignite/raft/jraft/test/MockAsyncContext.java
index 9de6606..83a1ce4 100644
--- a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/test/MockAsyncContext.java
+++ b/modules/raft/src/test/java/org/apache/ignite/raft/jraft/test/MockAsyncContext.java
@@ -22,7 +22,7 @@ import org.apache.ignite.raft.jraft.rpc.Message;
import org.apache.ignite.raft.jraft.rpc.RpcContext;
/**
- * mock alipay remoting async context
+ * Mocked async context.
*/
public class MockAsyncContext implements RpcContext {
private Object responseObject;
diff --git a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
index 42abedd..5f276d4 100644
--- a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
+++ b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
@@ -20,6 +20,7 @@ package org.apache.ignite.internal.app;
import org.apache.ignite.app.Ignite;
import org.apache.ignite.internal.vault.VaultManager;
import org.apache.ignite.table.manager.IgniteTables;
+import org.apache.ignite.tx.IgniteTransactions;
/**
* Ignite internal implementation.
@@ -46,6 +47,11 @@ public class IgniteImpl implements Ignite {
}
/** {@inheritDoc} */
+ @Override public IgniteTransactions transactions() {
+ return null;
+ }
+
+ /** {@inheritDoc} */
@Override public void close() throws Exception {
vaultManager.close();
}
diff --git a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ITDistributedTableTest.java b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ITDistributedTableTest.java
index edb54be..c073e30 100644
--- a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ITDistributedTableTest.java
+++ b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ITDistributedTableTest.java
@@ -265,8 +265,7 @@ public class ITDistributedTableTest {
@Override public Row resolve(BinaryRow row) {
return new Row(SCHEMA, row);
}
- },
- null);
+ }, null, null);
partitionedTableView(tbl, PARTS * 10);
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/AbstractTableView.java b/modules/table/src/main/java/org/apache/ignite/internal/table/AbstractTableView.java
index c5fe733..6536063 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/AbstractTableView.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/AbstractTableView.java
@@ -21,6 +21,8 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import org.apache.ignite.internal.schema.SchemaRegistry;
import org.apache.ignite.lang.IgniteInternalException;
+import org.apache.ignite.tx.Transaction;
+import org.jetbrains.annotations.Nullable;
/**
* Base class for Table views.
@@ -32,17 +34,35 @@ abstract class AbstractTableView {
/** Schema registry. */
protected final SchemaRegistry schemaReg;
+ /** The transaction */
+ protected final @Nullable Transaction tx;
+
/**
* Constructor
* @param tbl Internal table.
* @param schemaReg Schema registry.
+ * @param tx The transaction.
*/
- protected AbstractTableView(InternalTable tbl, SchemaRegistry schemaReg) {
+ protected AbstractTableView(InternalTable tbl, SchemaRegistry schemaReg, @Nullable Transaction tx) {
this.tbl = tbl;
this.schemaReg = schemaReg;
+ this.tx = tx;
+ }
+
+ /**
+ * @return Current transaction.
+ */
+ public @Nullable Transaction transaction() {
+ return tx;
}
/**
+ * @param tx The transaction.
+ * @return Transactional view.
+ */
+ public abstract AbstractTableView withTransaction(Transaction tx);
+
+ /**
* Waits for operation completion.
*
* @param fut Future to wait to.
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/InternalTable.java b/modules/table/src/main/java/org/apache/ignite/internal/table/InternalTable.java
index 05c8cce..cf1930e 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/InternalTable.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/InternalTable.java
@@ -22,7 +22,9 @@ import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.internal.schema.BinaryRow;
import org.apache.ignite.schema.SchemaMode;
+import org.apache.ignite.tx.Transaction;
import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
/**
* Internal table facade provides low-level methods for table operations.
@@ -59,123 +61,139 @@ public interface InternalTable {
* Asynchronously gets a row with same key columns values as given one from the table.
*
* @param keyRow Row with key columns set.
+ * @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- @NotNull CompletableFuture<BinaryRow> get(BinaryRow keyRow);
+ CompletableFuture<BinaryRow> get(BinaryRow keyRow, @Nullable Transaction tx);
/**
* Asynchronously get rows from the table.
*
* @param keyRows Rows with key columns set.
+ * @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- @NotNull CompletableFuture<Collection<BinaryRow>> getAll(Collection<BinaryRow> keyRows);
+ CompletableFuture<Collection<BinaryRow>> getAll(Collection<BinaryRow> keyRows, @Nullable Transaction tx);
/**
* Asynchronously inserts a row into the table if does not exist or replaces the existed one.
*
* @param row Row to insert into the table.
+ * @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- @NotNull CompletableFuture<Void> upsert(BinaryRow row);
+ CompletableFuture<Void> upsert(BinaryRow row, @Nullable Transaction tx);
/**
* Asynchronously inserts a row into the table if does not exist or replaces the existed one.
*
* @param rows Rows to insert into the table.
+ * @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- @NotNull CompletableFuture<Void> upsertAll(Collection<BinaryRow> rows);
+ CompletableFuture<Void> upsertAll(Collection<BinaryRow> rows, @Nullable Transaction tx);
/**
* Asynchronously inserts a row into the table or replaces if exists and return replaced previous row.
*
* @param row Row to insert into the table.
+ * @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- @NotNull CompletableFuture<BinaryRow> getAndUpsert(BinaryRow row);
+ CompletableFuture<BinaryRow> getAndUpsert(BinaryRow row, @Nullable Transaction tx);
/**
* Asynchronously inserts a row into the table if not exists.
*
* @param row Row to insert into the table.
+ * @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- @NotNull CompletableFuture<Boolean> insert(BinaryRow row);
+ CompletableFuture<Boolean> insert(BinaryRow row, @Nullable Transaction tx);
/**
* Asynchronously insert rows into the table which do not exist, skipping existed ones.
*
* @param rows Rows to insert into the table.
+ * @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- @NotNull CompletableFuture<Collection<BinaryRow>> insertAll(Collection<BinaryRow> rows);
+ CompletableFuture<Collection<BinaryRow>> insertAll(Collection<BinaryRow> rows, @Nullable Transaction tx);
/**
* Asynchronously replaces an existed row associated with the same key columns values as the given one has.
*
* @param row Row to replace with.
+ * @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- @NotNull CompletableFuture<Boolean> replace(BinaryRow row);
+ CompletableFuture<Boolean> replace(BinaryRow row, @Nullable Transaction tx);
/**
* Asynchronously replaces an expected row in the table with the given new one.
*
* @param oldRow Row to replace.
* @param newRow Row to replace with.
+ * @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- @NotNull CompletableFuture<Boolean> replace(BinaryRow oldRow, BinaryRow newRow);
+ CompletableFuture<Boolean> replace(BinaryRow oldRow, BinaryRow newRow, @Nullable Transaction tx);
/**
* Asynchronously gets an existed row associated with the same key columns values as the given one has,
* then replaces with the given one.
*
* @param row Row to replace with.
+ * @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- @NotNull CompletableFuture<BinaryRow> getAndReplace(BinaryRow row);
+ CompletableFuture<BinaryRow> getAndReplace(BinaryRow row, @Nullable Transaction tx);
/**
* Asynchronously deletes a row with the same key columns values as the given one from the table.
*
* @param keyRow Row with key columns set.
+ * @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- @NotNull CompletableFuture<Boolean> delete(BinaryRow keyRow);
+ CompletableFuture<Boolean> delete(BinaryRow keyRow, @Nullable Transaction tx);
/**
* Asynchronously deletes given row from the table.
*
* @param oldRow Row to delete.
+ * @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- @NotNull CompletableFuture<Boolean> deleteExact(BinaryRow oldRow);
+ CompletableFuture<Boolean> deleteExact(BinaryRow oldRow, @Nullable Transaction tx);
/**
* Asynchronously gets then deletes a row with the same key columns values from the table.
*
* @param row Row with key columns set.
+ * @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- @NotNull CompletableFuture<BinaryRow> getAndDelete(BinaryRow row);
+ CompletableFuture<BinaryRow> getAndDelete(BinaryRow row, @Nullable Transaction tx);
/**
* Asynchronously remove rows with the same key columns values as the given one has from the table.
*
* @param rows Rows with key columns set.
+ * @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- @NotNull CompletableFuture<Collection<BinaryRow>> deleteAll(Collection<BinaryRow> rows);
+ CompletableFuture<Collection<BinaryRow>> deleteAll(Collection<BinaryRow> rows, @Nullable Transaction tx);
/**
* Asynchronously remove given rows from the table.
*
* @param rows Rows to delete.
+ * @param tx The transaction.
* @return Future representing pending completion of the operation.
*/
- @NotNull CompletableFuture<Collection<BinaryRow>> deleteAllExact(Collection<BinaryRow> rows);
+ CompletableFuture<Collection<BinaryRow>> deleteAllExact(Collection<BinaryRow> rows,
+ @Nullable Transaction tx);
//TODO: IGNTIE-14488. Add invoke() methods.
}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/KVBinaryViewImpl.java b/modules/table/src/main/java/org/apache/ignite/internal/table/KVBinaryViewImpl.java
index 0ecb8a1..523ef4f 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/KVBinaryViewImpl.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/KVBinaryViewImpl.java
@@ -32,6 +32,7 @@ import org.apache.ignite.table.InvokeProcessor;
import org.apache.ignite.table.KeyValueBinaryView;
import org.apache.ignite.table.Tuple;
import org.apache.ignite.table.TupleBuilder;
+import org.apache.ignite.tx.Transaction;
import org.jetbrains.annotations.NotNull;
/**
@@ -46,12 +47,12 @@ public class KVBinaryViewImpl extends AbstractTableView implements KeyValueBinar
/**
* Constructor.
- *
* @param tbl Table storage.
* @param schemaReg Schema registry.
+ * @param tx The transaction.
*/
- public KVBinaryViewImpl(InternalTable tbl, SchemaRegistry schemaReg) {
- super(tbl, schemaReg);
+ public KVBinaryViewImpl(InternalTable tbl, SchemaRegistry schemaReg, Transaction tx) {
+ super(tbl, schemaReg, tx);
marsh = new TupleMarshallerImpl(schemaReg);
@@ -64,8 +65,8 @@ public class KVBinaryViewImpl extends AbstractTableView implements KeyValueBinar
* @param tbl Table storage.
* @param schemaReg Schema registry.
*/
- public KVBinaryViewImpl(InternalTable tbl, SchemaRegistry schemaReg, TableManager tblMgr) {
- super(tbl, schemaReg);
+ public KVBinaryViewImpl(InternalTable tbl, SchemaRegistry schemaReg, TableManager tblMgr, Transaction tx) {
+ super(tbl, schemaReg, tx);
marsh = new TupleMarshallerImpl(schemaReg);
@@ -83,7 +84,7 @@ public class KVBinaryViewImpl extends AbstractTableView implements KeyValueBinar
Row kRow = marshaller().marshal(key, null); // Convert to portable format to pass TX/storage layer.
- return tbl.get(kRow) // Load async.
+ return tbl.get(kRow, tx) // Load async.
.thenApply(this::wrap) // Binary -> schema-aware row
.thenApply(t -> t == null ? null : t.valueChunk()); // Narrow to value.
}
@@ -97,7 +98,7 @@ public class KVBinaryViewImpl extends AbstractTableView implements KeyValueBinar
@Override public @NotNull CompletableFuture<Map<Tuple, Tuple>> getAllAsync(@NotNull Collection<Tuple> keys) {
Objects.requireNonNull(keys);
- return tbl.getAll(keys.stream().map(k -> marsh.marshal(k, null)).collect(Collectors.toList()))
+ return tbl.getAll(keys.stream().map(k -> marsh.marshal(k, null)).collect(Collectors.toList()), tx)
.thenApply(this::wrap)
.thenApply(ts -> ts.stream().filter(Objects::nonNull).collect(Collectors.toMap(TableRow::keyChunk, TableRow::valueChunk)));
}
@@ -118,7 +119,7 @@ public class KVBinaryViewImpl extends AbstractTableView implements KeyValueBinar
Row row = marshaller().marshal(key, val); // Convert to portable format to pass TX/storage layer.
- return tbl.upsert(row);
+ return tbl.upsert(row, tx);
}
/** {@inheritDoc} */
@@ -133,7 +134,7 @@ public class KVBinaryViewImpl extends AbstractTableView implements KeyValueBinar
return tbl.upsertAll(pairs.entrySet()
.stream()
.map(this::marshalPair)
- .collect(Collectors.toList()));
+ .collect(Collectors.toList()), tx);
}
/** {@inheritDoc} */
@@ -147,7 +148,7 @@ public class KVBinaryViewImpl extends AbstractTableView implements KeyValueBinar
Row row = marshaller().marshal(key, val); // Convert to portable format to pass TX/storage layer.
- return tbl.getAndUpsert(row)
+ return tbl.getAndUpsert(row, tx)
.thenApply(this::wrap) // Binary -> schema-aware row
.thenApply(t -> t == null ? null : t.valueChunk()); // Narrow to value.
}
@@ -163,7 +164,7 @@ public class KVBinaryViewImpl extends AbstractTableView implements KeyValueBinar
Row row = marshaller().marshal(key, val); // Convert to portable format to pass TX/storage layer.
- return tbl.insert(row);
+ return tbl.insert(row, tx);
}
/** {@inheritDoc} */
@@ -177,7 +178,7 @@ public class KVBinaryViewImpl extends AbstractTableView implements KeyValueBinar
Row row = marshaller().marshal(key, null); // Convert to portable format to pass TX/storage layer.
- return tbl.delete(row);
+ return tbl.delete(row, tx);
}
/** {@inheritDoc} */
@@ -192,7 +193,7 @@ public class KVBinaryViewImpl extends AbstractTableView implements KeyValueBinar
Row row = marshaller().marshal(key, val); // Convert to portable format to pass TX/storage layer.
- return tbl.deleteExact(row);
+ return tbl.deleteExact(row, tx);
}
/** {@inheritDoc} */
@@ -206,7 +207,7 @@ public class KVBinaryViewImpl extends AbstractTableView implements KeyValueBinar
@Override public @NotNull CompletableFuture<Collection<Tuple>> removeAllAsync(@NotNull Collection<Tuple> keys) {
Objects.requireNonNull(keys);
- return tbl.deleteAll(keys.stream().map(k -> marsh.marshal(k, null)).collect(Collectors.toList()))
+ return tbl.deleteAll(keys.stream().map(k -> marsh.marshal(k, null)).collect(Collectors.toList()), tx)
.thenApply(this::wrap)
.thenApply(t -> t.stream().filter(Objects::nonNull).map(TableRow::valueChunk).collect(Collectors.toList()));
}
@@ -222,7 +223,7 @@ public class KVBinaryViewImpl extends AbstractTableView implements KeyValueBinar
@Override public @NotNull CompletableFuture<Tuple> getAndRemoveAsync(@NotNull Tuple key) {
Objects.requireNonNull(key);
- return tbl.getAndDelete(marsh.marshal(key, null))
+ return tbl.getAndDelete(marsh.marshal(key, null), tx)
.thenApply(this::wrap)
.thenApply(t -> t == null ? null : t.valueChunk());
}
@@ -238,7 +239,7 @@ public class KVBinaryViewImpl extends AbstractTableView implements KeyValueBinar
Row row = marshaller().marshal(key, val); // Convert to portable format to pass TX/storage layer.
- return tbl.replace(row);
+ return tbl.replace(row, tx);
}
/** {@inheritDoc} */
@@ -253,7 +254,7 @@ public class KVBinaryViewImpl extends AbstractTableView implements KeyValueBinar
Row oldRow = marshaller().marshal(key, oldVal); // Convert to portable format to pass TX/storage layer.
Row newRow = marshaller().marshal(key, newVal); // Convert to portable format to pass TX/storage layer.
- return tbl.replace(oldRow, newRow);
+ return tbl.replace(oldRow, newRow, tx);
}
/** {@inheritDoc} */
@@ -265,7 +266,7 @@ public class KVBinaryViewImpl extends AbstractTableView implements KeyValueBinar
@Override public @NotNull CompletableFuture<Tuple> getAndReplaceAsync(@NotNull Tuple key, Tuple val) {
Objects.requireNonNull(key);
- return tbl.getAndReplace(marsh.marshal(key, val))
+ return tbl.getAndReplace(marsh.marshal(key, val), tx)
.thenApply(this::wrap)
.thenApply(t -> t == null ? null : t.valueChunk());
}
@@ -318,6 +319,11 @@ public class KVBinaryViewImpl extends AbstractTableView implements KeyValueBinar
throw new IllegalArgumentException("Unknown schema type: " + tbl.schemaMode());
}
+ /** {@inheritDoc} */
+ @Override public KVBinaryViewImpl withTransaction(Transaction tx) {
+ return new KVBinaryViewImpl(tbl, schemaReg, tx);
+ }
+
/**
* @return Marshaller.
*/
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/KVViewImpl.java b/modules/table/src/main/java/org/apache/ignite/internal/table/KVViewImpl.java
index c744e93..614c99a 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/KVViewImpl.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/KVViewImpl.java
@@ -31,7 +31,9 @@ import org.apache.ignite.table.InvokeProcessor;
import org.apache.ignite.table.KeyValueView;
import org.apache.ignite.table.mapper.KeyMapper;
import org.apache.ignite.table.mapper.ValueMapper;
+import org.apache.ignite.tx.Transaction;
import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
/**
* Key-value view implementation.
@@ -39,15 +41,15 @@ import org.jetbrains.annotations.NotNull;
public class KVViewImpl<K, V> extends AbstractTableView implements KeyValueView<K, V> {
/**
* Constructor.
- *
* @param tbl Table storage.
* @param schemaReg Schema registry.
* @param keyMapper Key class mapper.
* @param valueMapper Value class mapper.
+ * @param tx The transaction.
*/
public KVViewImpl(InternalTable tbl, SchemaRegistry schemaReg, KeyMapper<K> keyMapper,
- ValueMapper<V> valueMapper) {
- super(tbl, schemaReg);
+ ValueMapper<V> valueMapper, @Nullable Transaction tx) {
+ super(tbl, schemaReg, tx);
}
/** {@inheritDoc} */
@@ -63,7 +65,7 @@ public class KVViewImpl<K, V> extends AbstractTableView implements KeyValueView<
Row kRow = marsh.serialize(key, null); // Convert to portable format to pass TX/storage layer.
- return tbl.get(kRow)
+ return tbl.get(kRow, tx)
.thenApply(this::wrap) // Binary -> schema-aware row
.thenApply(marsh::deserializeValue); // row -> deserialized obj.
}
@@ -224,6 +226,11 @@ public class KVViewImpl<K, V> extends AbstractTableView implements KeyValueView<
throw new UnsupportedOperationException("Not implemented yet.");
}
+ /** {@inheritDoc} */
+ @Override public KVViewImpl<K, V> withTransaction(Transaction tx) {
+ throw new UnsupportedOperationException("Not implemented yet.");
+ }
+
/**
* @return Marshaller.
*/
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/RecordViewImpl.java b/modules/table/src/main/java/org/apache/ignite/internal/table/RecordViewImpl.java
index 27bdc44..129a775 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/RecordViewImpl.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/RecordViewImpl.java
@@ -30,7 +30,9 @@ import org.apache.ignite.internal.schema.SchemaRegistry;
import org.apache.ignite.table.InvokeProcessor;
import org.apache.ignite.table.RecordView;
import org.apache.ignite.table.mapper.RecordMapper;
+import org.apache.ignite.tx.Transaction;
import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
/**
* Record view implementation.
@@ -38,13 +40,13 @@ import org.jetbrains.annotations.NotNull;
public class RecordViewImpl<R> extends AbstractTableView implements RecordView<R> {
/**
* Constructor.
- *
- * @param tbl Table.
+ * @param tbl Table.
* @param schemaReg Schema registry.
* @param mapper Record class mapper.
+ * @param tx The transaction.
*/
- public RecordViewImpl(InternalTable tbl, SchemaRegistry schemaReg, RecordMapper<R> mapper) {
- super(tbl, schemaReg);
+ public RecordViewImpl(InternalTable tbl, SchemaRegistry schemaReg, RecordMapper<R> mapper, @Nullable Transaction tx) {
+ super(tbl, schemaReg, tx);
}
/** {@inheritDoc} */
@@ -60,7 +62,7 @@ public class RecordViewImpl<R> extends AbstractTableView implements RecordView<R
Row kRow = marsh.serialize(recObjToFill); // Convert to portable format to pass TX/storage layer.
- return tbl.get(kRow) // Load async.
+ return tbl.get(kRow, tx) // Load async.
.thenApply(this::wrap) // Binary -> schema-aware row
.thenApply(r -> marsh.deserialize(r, recObjToFill)); // Deserialize and fill record.
}
@@ -78,7 +80,7 @@ public class RecordViewImpl<R> extends AbstractTableView implements RecordView<R
Row kRow = marsh.serialize(keyRec); // Convert to portable format to pass TX/storage layer.
- return tbl.get(kRow) // Load async.
+ return tbl.get(kRow, tx) // Load async.
.thenApply(this::wrap) // Binary -> schema-aware row
.thenApply(marsh::deserialize); // Deserialize.
}
@@ -252,6 +254,11 @@ public class RecordViewImpl<R> extends AbstractTableView implements RecordView<R
throw new UnsupportedOperationException("Not implemented yet.");
}
+ /** {@inheritDoc} */
+ @Override public RecordViewImpl<R> withTransaction(Transaction tx) {
+ throw new UnsupportedOperationException("Not implemented yet.");
+ }
+
/**
* @return Marshaller.
*/
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/TableImpl.java b/modules/table/src/main/java/org/apache/ignite/internal/table/TableImpl.java
index 960e420..caaa41d 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/TableImpl.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/TableImpl.java
@@ -41,7 +41,9 @@ import org.apache.ignite.table.TupleBuilder;
import org.apache.ignite.table.mapper.KeyMapper;
import org.apache.ignite.table.mapper.RecordMapper;
import org.apache.ignite.table.mapper.ValueMapper;
+import org.apache.ignite.tx.Transaction;
import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
/**
* Table view implementation for binary objects.
@@ -58,9 +60,11 @@ public class TableImpl extends AbstractTableView implements Table {
*
* @param tbl Table.
* @param schemaReg Table schema registry.
+ * @param tblMgr Table manager.
+ * @param tx The transaction.
*/
- public TableImpl(InternalTable tbl, SchemaRegistry schemaReg, TableManager tblMgr) {
- super(tbl, schemaReg);
+ public TableImpl(InternalTable tbl, SchemaRegistry schemaReg, TableManager tblMgr, @Nullable Transaction tx) {
+ super(tbl, schemaReg, tx);
marsh = new TupleMarshallerImpl(schemaReg);
@@ -92,17 +96,22 @@ public class TableImpl extends AbstractTableView implements Table {
/** {@inheritDoc} */
@Override public <R> RecordView<R> recordView(RecordMapper<R> recMapper) {
- return new RecordViewImpl<>(tbl, schemaReg, recMapper);
+ return new RecordViewImpl<>(tbl, schemaReg, recMapper, tx);
}
/** {@inheritDoc} */
@Override public <K, V> KeyValueView<K, V> kvView(KeyMapper<K> keyMapper, ValueMapper<V> valMapper) {
- return new KVViewImpl<>(tbl, schemaReg, keyMapper, valMapper);
+ return new KVViewImpl<>(tbl, schemaReg, keyMapper, valMapper, tx);
}
/** {@inheritDoc} */
@Override public KeyValueBinaryView kvView() {
- return new KVBinaryViewImpl(tbl, schemaReg, tblMgr);
+ return new KVBinaryViewImpl(tbl, schemaReg, tblMgr, tx);
+ }
+
+ /** {@inheritDoc} */
+ @Override public TableImpl withTransaction(Transaction tx) {
+ return new TableImpl(tbl, schemaReg, tblMgr, tx);
}
/** {@inheritDoc} */
@@ -116,7 +125,7 @@ public class TableImpl extends AbstractTableView implements Table {
final Row keyRow = marshaller().marshal(keyRec, null); // Convert to portable format to pass TX/storage layer.
- return tbl.get(keyRow).thenApply(this::wrap);
+ return tbl.get(keyRow, tx).thenApply(this::wrap);
}
/** {@inheritDoc} */
@@ -136,7 +145,7 @@ public class TableImpl extends AbstractTableView implements Table {
keys.add(keyRow);
}
- return tbl.getAll(keys).thenApply(this::wrap);
+ return tbl.getAll(keys, tx).thenApply(this::wrap);
}
/** {@inheritDoc} */
@@ -150,7 +159,7 @@ public class TableImpl extends AbstractTableView implements Table {
final Row keyRow = marshaller().marshal(rec);
- return tbl.upsert(keyRow);
+ return tbl.upsert(keyRow, tx);
}
/** {@inheritDoc} */
@@ -170,7 +179,7 @@ public class TableImpl extends AbstractTableView implements Table {
keys.add(keyRow);
}
- return tbl.upsertAll(keys);
+ return tbl.upsertAll(keys, tx);
}
/** {@inheritDoc} */
@@ -184,7 +193,7 @@ public class TableImpl extends AbstractTableView implements Table {
final Row keyRow = marshaller().marshal(rec);
- return tbl.getAndUpsert(keyRow).thenApply(this::wrap);
+ return tbl.getAndUpsert(keyRow, tx).thenApply(this::wrap);
}
/** {@inheritDoc} */
@@ -198,7 +207,7 @@ public class TableImpl extends AbstractTableView implements Table {
final Row keyRow = marshaller().marshal(rec);
- return tbl.insert(keyRow);
+ return tbl.insert(keyRow, tx);
}
/** {@inheritDoc} */
@@ -218,7 +227,7 @@ public class TableImpl extends AbstractTableView implements Table {
keys.add(keyRow);
}
- return tbl.insertAll(keys).thenApply(this::wrap);
+ return tbl.insertAll(keys, tx).thenApply(this::wrap);
}
/** {@inheritDoc} */
@@ -232,7 +241,7 @@ public class TableImpl extends AbstractTableView implements Table {
final Row keyRow = marshaller().marshal(rec);
- return tbl.replace(keyRow);
+ return tbl.replace(keyRow, tx);
}
/** {@inheritDoc} */
@@ -248,7 +257,7 @@ public class TableImpl extends AbstractTableView implements Table {
final Row oldRow = marshaller().marshal(oldRec);
final Row newRow = marshaller().marshal(newRec);
- return tbl.replace(oldRow, newRow);
+ return tbl.replace(oldRow, newRow, tx);
}
/** {@inheritDoc} */
@@ -262,7 +271,7 @@ public class TableImpl extends AbstractTableView implements Table {
final Row keyRow = marshaller().marshal(rec);
- return tbl.getAndReplace(keyRow).thenApply(this::wrap);
+ return tbl.getAndReplace(keyRow, tx).thenApply(this::wrap);
}
/** {@inheritDoc} */
@@ -276,7 +285,7 @@ public class TableImpl extends AbstractTableView implements Table {
final Row keyRow = marshaller().marshal(keyRec, null);
- return tbl.delete(keyRow);
+ return tbl.delete(keyRow, tx);
}
/** {@inheritDoc} */
@@ -290,7 +299,7 @@ public class TableImpl extends AbstractTableView implements Table {
final Row row = marshaller().marshal(rec);
- return tbl.deleteExact(row);
+ return tbl.deleteExact(row, tx);
}
/** {@inheritDoc} */
@@ -304,7 +313,7 @@ public class TableImpl extends AbstractTableView implements Table {
final Row row = marshaller().marshal(rec);
- return tbl.getAndDelete(row).thenApply(this::wrap);
+ return tbl.getAndDelete(row, tx).thenApply(this::wrap);
}
/** {@inheritDoc} */
@@ -324,7 +333,7 @@ public class TableImpl extends AbstractTableView implements Table {
keys.add(keyRow);
}
- return tbl.deleteAll(keys).thenApply(this::wrap);
+ return tbl.deleteAll(keys, tx).thenApply(this::wrap);
}
/** {@inheritDoc} */
@@ -346,7 +355,7 @@ public class TableImpl extends AbstractTableView implements Table {
keys.add(keyRow);
}
- return tbl.deleteAllExact(keys).thenApply(this::wrap);
+ return tbl.deleteAllExact(keys, tx).thenApply(this::wrap);
}
/** {@inheritDoc} */
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
index a3d2ee7..939fd07 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
@@ -164,7 +164,7 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
InternalTableImpl internalTable = new InternalTableImpl(name, tblId, partitionMap, partitions);
- var table = new TableImpl(internalTable, schemaReg, this);
+ var table = new TableImpl(internalTable, schemaReg, this, null);
tables.put(name, table);
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
index acb0efb..e6c10c3 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
@@ -45,6 +45,7 @@ import org.apache.ignite.internal.table.distributed.command.response.MultiRowsRe
import org.apache.ignite.internal.table.distributed.command.response.SingleRowResponse;
import org.apache.ignite.raft.client.service.RaftGroupService;
import org.apache.ignite.schema.SchemaMode;
+import org.apache.ignite.tx.Transaction;
import org.jetbrains.annotations.NotNull;
/**
@@ -107,13 +108,14 @@ public class InternalTableImpl implements InternalTable {
}
/** {@inheritDoc} */
- @Override public @NotNull CompletableFuture<BinaryRow> get(BinaryRow keyRow) {
+ @Override public CompletableFuture<BinaryRow> get(BinaryRow keyRow, Transaction tx) {
return partitionMap.get(partId(keyRow)).<SingleRowResponse>run(new GetCommand(keyRow))
.thenApply(SingleRowResponse::getValue);
}
/** {@inheritDoc} */
- @Override public @NotNull CompletableFuture<Collection<BinaryRow>> getAll(Collection<BinaryRow> keyRows) {
+ @Override public CompletableFuture<Collection<BinaryRow>> getAll(Collection<BinaryRow> keyRows,
+ Transaction tx) {
HashMap<Integer, HashSet<BinaryRow>> keyRowsByPartition = new HashMap<>();
for (BinaryRow keyRow : keyRows) {
@@ -140,12 +142,12 @@ public class InternalTableImpl implements InternalTable {
}
/** {@inheritDoc} */
- @Override public @NotNull CompletableFuture<Void> upsert(BinaryRow row) {
+ @Override public CompletableFuture<Void> upsert(BinaryRow row, Transaction tx) {
return partitionMap.get(partId(row)).run(new UpsertCommand(row));
}
/** {@inheritDoc} */
- @Override public @NotNull CompletableFuture<Void> upsertAll(Collection<BinaryRow> rows) {
+ @Override public CompletableFuture<Void> upsertAll(Collection<BinaryRow> rows, Transaction tx) {
HashMap<Integer, HashSet<BinaryRow>> keyRowsByPartition = new HashMap<>();
for (BinaryRow keyRow : rows) {
@@ -167,18 +169,18 @@ public class InternalTableImpl implements InternalTable {
}
/** {@inheritDoc} */
- @Override public @NotNull CompletableFuture<BinaryRow> getAndUpsert(BinaryRow row) {
+ @Override public CompletableFuture<BinaryRow> getAndUpsert(BinaryRow row, Transaction tx) {
return partitionMap.get(partId(row)).<SingleRowResponse>run(new GetAndUpsertCommand(row))
.thenApply(SingleRowResponse::getValue);
}
/** {@inheritDoc} */
- @Override public @NotNull CompletableFuture<Boolean> insert(BinaryRow row) {
+ @Override public CompletableFuture<Boolean> insert(BinaryRow row, Transaction tx) {
return partitionMap.get(partId(row)).run(new InsertCommand(row));
}
/** {@inheritDoc} */
- @Override public @NotNull CompletableFuture<Collection<BinaryRow>> insertAll(Collection<BinaryRow> rows) {
+ @Override public CompletableFuture<Collection<BinaryRow>> insertAll(Collection<BinaryRow> rows, Transaction tx) {
HashMap<Integer, HashSet<BinaryRow>> keyRowsByPartition = new HashMap<>();
for (BinaryRow keyRow : rows) {
@@ -205,39 +207,41 @@ public class InternalTableImpl implements InternalTable {
}
/** {@inheritDoc} */
- @Override public @NotNull CompletableFuture<Boolean> replace(BinaryRow row) {
+ @Override public CompletableFuture<Boolean> replace(BinaryRow row, Transaction tx) {
return partitionMap.get(partId(row)).<Boolean>run(new ReplaceIfExistCommand(row));
}
/** {@inheritDoc} */
- @Override public @NotNull CompletableFuture<Boolean> replace(BinaryRow oldRow, BinaryRow newRow) {
+ @Override public CompletableFuture<Boolean> replace(BinaryRow oldRow, BinaryRow newRow,
+ Transaction tx) {
return partitionMap.get(partId(oldRow)).run(new ReplaceCommand(oldRow, newRow));
}
/** {@inheritDoc} */
- @Override public @NotNull CompletableFuture<BinaryRow> getAndReplace(BinaryRow row) {
+ @Override public CompletableFuture<BinaryRow> getAndReplace(BinaryRow row, Transaction tx) {
return partitionMap.get(partId(row)).<SingleRowResponse>run(new ReplaceIfExistCommand(row))
.thenApply(SingleRowResponse::getValue);
}
/** {@inheritDoc} */
- @Override public @NotNull CompletableFuture<Boolean> delete(BinaryRow keyRow) {
+ @Override public CompletableFuture<Boolean> delete(BinaryRow keyRow, Transaction tx) {
return partitionMap.get(partId(keyRow)).run(new DeleteCommand(keyRow));
}
/** {@inheritDoc} */
- @Override public @NotNull CompletableFuture<Boolean> deleteExact(BinaryRow oldRow) {
+ @Override public CompletableFuture<Boolean> deleteExact(BinaryRow oldRow, Transaction tx) {
return partitionMap.get(partId(oldRow)).<Boolean>run(new DeleteExactCommand(oldRow));
}
/** {@inheritDoc} */
- @Override public @NotNull CompletableFuture<BinaryRow> getAndDelete(BinaryRow row) {
+ @Override public CompletableFuture<BinaryRow> getAndDelete(BinaryRow row, Transaction tx) {
return partitionMap.get(partId(row)).<SingleRowResponse>run(new GetAndDeleteCommand(row))
.thenApply(SingleRowResponse::getValue);
}
/** {@inheritDoc} */
- @Override public @NotNull CompletableFuture<Collection<BinaryRow>> deleteAll(Collection<BinaryRow> rows) {
+ @Override public CompletableFuture<Collection<BinaryRow>> deleteAll(Collection<BinaryRow> rows,
+ Transaction tx) {
HashMap<Integer, HashSet<BinaryRow>> keyRowsByPartition = new HashMap<>();
for (BinaryRow keyRow : rows) {
@@ -264,7 +268,8 @@ public class InternalTableImpl implements InternalTable {
}
/** {@inheritDoc} */
- @Override public @NotNull CompletableFuture<Collection<BinaryRow>> deleteAllExact(Collection<BinaryRow> rows) {
+ @Override public CompletableFuture<Collection<BinaryRow>> deleteAllExact(Collection<BinaryRow> rows,
+ Transaction tx) {
HashMap<Integer, HashSet<BinaryRow>> keyRowsByPartition = new HashMap<>();
for (BinaryRow keyRow : rows) {
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/Example.java b/modules/table/src/test/java/org/apache/ignite/internal/table/Example.java
index 13c9a0c..9836e12 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/Example.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/Example.java
@@ -43,7 +43,7 @@ public class Example {
* @return Table implementation.
*/
private static List<Table> tableFactory() {
- return Collections.singletonList(new TableImpl(new DummyInternalTableImpl(), null, null));
+ return Collections.singletonList(new TableImpl(new DummyInternalTableImpl(), null, null, null));
}
/**
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/KVViewOperationsTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/KVViewOperationsTest.java
index fec7740..2f11d0e 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/KVViewOperationsTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/KVViewOperationsTest.java
@@ -56,7 +56,8 @@ public class KVViewOperationsTest {
new Column[] {new Column("val", NativeTypes.INT64, false)}
);
- KeyValueBinaryView tbl = new KVBinaryViewImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema));
+ KeyValueBinaryView tbl =
+ new KVBinaryViewImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null);
final Tuple key = tbl.tupleBuilder().set("id", 1L).build();
final Tuple val = tbl.tupleBuilder().set("val", 11L).build();
@@ -99,7 +100,8 @@ public class KVViewOperationsTest {
new Column[] {new Column("val", NativeTypes.INT64, false)}
);
- KeyValueBinaryView tbl = new KVBinaryViewImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema));
+ KeyValueBinaryView tbl =
+ new KVBinaryViewImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null);
final Tuple key = tbl.tupleBuilder().set("id", 1L).build();
final Tuple val = tbl.tupleBuilder().set("val", 11L).build();
@@ -132,7 +134,8 @@ public class KVViewOperationsTest {
new Column[] {new Column("val", NativeTypes.INT64, false)}
);
- KeyValueBinaryView tbl = new KVBinaryViewImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema));
+ KeyValueBinaryView tbl =
+ new KVBinaryViewImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null);
final Tuple key = tbl.tupleBuilder().set("id", 1L).build();
final Tuple val = tbl.tupleBuilder().set("val", 11L).build();
@@ -167,7 +170,8 @@ public class KVViewOperationsTest {
new Column[] {new Column("val", NativeTypes.INT64, false)}
);
- KeyValueBinaryView tbl = new KVBinaryViewImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema));
+ KeyValueBinaryView tbl =
+ new KVBinaryViewImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null);
final Tuple key = tbl.tupleBuilder().set("id", 1L).build();
final Tuple key2 = tbl.tupleBuilder().set("id", 2L).build();
@@ -210,7 +214,8 @@ public class KVViewOperationsTest {
new Column[] {new Column("val", NativeTypes.INT64, false)}
);
- KeyValueBinaryView tbl = new KVBinaryViewImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema));
+ KeyValueBinaryView tbl =
+ new KVBinaryViewImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null);
final Tuple key = tbl.tupleBuilder().set("id", 1L).build();
final Tuple key2 = tbl.tupleBuilder().set("id", 2L).build();
@@ -267,7 +272,8 @@ public class KVViewOperationsTest {
new Column[] {new Column("val", NativeTypes.INT64, false)}
);
- KeyValueBinaryView tbl = new KVBinaryViewImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema));
+ KeyValueBinaryView tbl =
+ new KVBinaryViewImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null);
final Tuple key = tbl.tupleBuilder().set("id", 1L).build();
final Tuple key2 = tbl.tupleBuilder().set("id", 2L).build();
@@ -313,7 +319,8 @@ public class KVViewOperationsTest {
new Column[] {new Column("val", NativeTypes.INT64, false)}
);
- KeyValueBinaryView tbl = new KVBinaryViewImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema));
+ KeyValueBinaryView tbl =
+ new KVBinaryViewImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null);
final Tuple key = tbl.tupleBuilder().set("id", 1L).build();
final Tuple key2 = tbl.tupleBuilder().set("id", 2L).build();
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/StrictSchemaOperationsTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/StrictSchemaOperationsTest.java
index 22bad4c..ec79bb6 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/StrictSchemaOperationsTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/StrictSchemaOperationsTest.java
@@ -47,7 +47,7 @@ public class StrictSchemaOperationsTest {
new Column[] {new Column("val", NativeTypes.INT64, false)}
);
- Table tbl = new TableImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null);
+ Table tbl = new TableImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null, null);
assertThrows(ColumnNotFoundException.class, () -> tbl.tupleBuilder().set("invalidCol", 0));
}
@@ -67,7 +67,7 @@ public class StrictSchemaOperationsTest {
}
);
- Table tbl = new TableImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null);
+ Table tbl = new TableImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null, null);
// Check not-nullable column.
assertThrows(IllegalArgumentException.class, () -> tbl.tupleBuilder().set("id", null));
@@ -93,7 +93,7 @@ public class StrictSchemaOperationsTest {
}
);
- Table tbl = new TableImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null);
+ Table tbl = new TableImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null, null);
tbl.tupleBuilder().set("valString", "qwe");
tbl.tupleBuilder().set("valString", "qw");
@@ -119,7 +119,7 @@ public class StrictSchemaOperationsTest {
new Column("valLimited", NativeTypes.blobOf(2), true)
});
- Table tbl = new TableImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null);
+ Table tbl = new TableImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null, null);
tbl.tupleBuilder().set("valUnlimited", null);
tbl.tupleBuilder().set("valLimited", null);
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/TableBinaryViewOperationsTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/TableBinaryViewOperationsTest.java
index f24babd..38007dc 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/TableBinaryViewOperationsTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/TableBinaryViewOperationsTest.java
@@ -57,7 +57,7 @@ public class TableBinaryViewOperationsTest {
new Column[]{new Column("val", NativeTypes.INT64, false)}
);
- Table tbl = new TableImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null);
+ Table tbl = new TableImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null, null);
final Tuple tuple = tbl.tupleBuilder().set("id", 1L).set("val", 11L).build();
final Tuple newTuple = tbl.tupleBuilder().set("id", 1L).set("val", 22L).build();
@@ -92,7 +92,7 @@ public class TableBinaryViewOperationsTest {
new Column[]{new Column("val", NativeTypes.INT64, false)}
);
- Table tbl = new TableImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null);
+ Table tbl = new TableImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null, null);
final Tuple tuple = tbl.tupleBuilder().set("id", 1L).set("val", 11L).build();
final Tuple newTuple = tbl.tupleBuilder().set("id", 1L).set("val", 22L).build();
@@ -128,7 +128,7 @@ public class TableBinaryViewOperationsTest {
new Column[]{new Column("val", NativeTypes.INT64, false)}
);
- Table tbl = new TableImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null);
+ Table tbl = new TableImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null, null);
final Tuple tuple = tbl.tupleBuilder().set("id", 1L).set("val", 11L).build();
final Tuple newTuple = tbl.tupleBuilder().set("id", 1L).set("val", 22L).build();
@@ -161,7 +161,7 @@ public class TableBinaryViewOperationsTest {
new Column[]{new Column("val", NativeTypes.INT64, false)}
);
- Table tbl = new TableImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null);
+ Table tbl = new TableImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null, null);
final Tuple tuple = tbl.tupleBuilder().set("id", 1L).set("val", 11L).build();
@@ -192,7 +192,7 @@ public class TableBinaryViewOperationsTest {
new Column[]{new Column("val", NativeTypes.INT64, false)}
);
- Table tbl = new TableImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null);
+ Table tbl = new TableImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null, null);
final Tuple keyTuple = tbl.tupleBuilder().set("id", 1L).build();
final Tuple tuple = tbl.tupleBuilder().set("id", 1L).set("val", 11L).build();
@@ -244,7 +244,7 @@ public class TableBinaryViewOperationsTest {
new Column[]{new Column("val", NativeTypes.INT64, false)}
);
- Table tbl = new TableImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null);
+ Table tbl = new TableImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null, null);
final Tuple keyTuple = tbl.tupleBuilder().set("id", 1L).build();
final Tuple tuple = tbl.tupleBuilder().set("id", 1L).set("val", 11L).build();
@@ -280,7 +280,7 @@ public class TableBinaryViewOperationsTest {
new Column[]{new Column("val", NativeTypes.INT64, false)}
);
- Table tbl = new TableImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null);
+ Table tbl = new TableImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null, null);
final Tuple keyTuple = tbl.tupleBuilder().set("id", 1L).build();
final Tuple tuple = tbl.tupleBuilder().set("id", 1L).set("val", 11L).build();
@@ -321,7 +321,7 @@ public class TableBinaryViewOperationsTest {
}
);
- Table tbl = new TableImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null);
+ Table tbl = new TableImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null, null);
final Tuple keyTuple0 = new TestTupleBuilder().set("id", 0).set("id1", 0).build();
final Tuple keyTuple1 = new TestTupleBuilder().set("id1", 0).build();
@@ -357,7 +357,7 @@ public class TableBinaryViewOperationsTest {
}
);
- Table tbl = new TableImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null);
+ Table tbl = new TableImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null, null);
final Tuple keyTuple0 = tbl.tupleBuilder().set("id", 0L).build();
final Tuple keyTuple1 = tbl.tupleBuilder().set("id", 1L).build();
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/TxTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/TxTest.java
new file mode 100644
index 0000000..53bbc92
--- /dev/null
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/TxTest.java
@@ -0,0 +1,208 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.table;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
+import org.apache.ignite.internal.schema.Column;
+import org.apache.ignite.internal.schema.NativeTypes;
+import org.apache.ignite.internal.schema.SchemaDescriptor;
+import org.apache.ignite.internal.table.impl.DummyInternalTableImpl;
+import org.apache.ignite.internal.table.impl.DummySchemaManagerImpl;
+import org.apache.ignite.internal.util.Pair;
+import org.apache.ignite.table.KeyValueBinaryView;
+import org.apache.ignite.table.Table;
+import org.apache.ignite.table.Tuple;
+import org.apache.ignite.tx.IgniteTransactions;
+import org.apache.ignite.tx.Transaction;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
+
+import static java.util.concurrent.CompletableFuture.allOf;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/** */
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.LENIENT)
+public class TxTest {
+ /** Table ID test value. */
+ public final java.util.UUID tableId = java.util.UUID.randomUUID();
+
+ /** Accounts table. */
+ private Table accounts;
+
+ /** */
+ public static final double BALANCE_1 = 500;
+
+ /** */
+ public static final double BALANCE_2 = 500;
+
+ /** */
+ public static final double DELTA = 500;
+
+ /** */
+ @Mock
+ private IgniteTransactions igniteTransactions;
+
+ /** */
+ @Mock
+ private Transaction tx;
+
+ /**
+ * Initialize the test state.
+ */
+ @BeforeEach
+ public void before() {
+ SchemaDescriptor schema = new SchemaDescriptor(
+ tableId,
+ 1,
+ new Column[]{new Column("accountNumber", NativeTypes.INT64, false)},
+ new Column[]{new Column("balance", NativeTypes.DOUBLE, false)}
+ );
+
+ accounts = new TableImpl(new DummyInternalTableImpl(), new DummySchemaManagerImpl(schema), null, null);
+ Tuple r1 = accounts.tupleBuilder().set("accountNumber", 1L).set("balance", BALANCE_1).build();
+ Tuple r2 = accounts.tupleBuilder().set("accountNumber", 2L).set("balance", BALANCE_2).build();
+
+ accounts.insert(r1);
+ accounts.insert(r2);
+
+ Mockito.doAnswer(invocation -> {
+ Consumer<Transaction> argument = invocation.getArgument(0);
+
+ argument.accept(tx);
+
+ return null;
+ }).when(igniteTransactions).runInTransaction(Mockito.any());
+
+ Mockito.when(igniteTransactions.beginAsync()).thenReturn(CompletableFuture.completedFuture(tx));
+ }
+
+ /**
+ * Tests a synchronous transaction.
+ */
+ @Test
+ public void testTxSync() {
+ igniteTransactions.runInTransaction(tx -> {
+ Table txAcc = accounts.withTransaction(tx);
+
+ CompletableFuture<Tuple> read1 = txAcc.getAsync(makeKey(1));
+ CompletableFuture<Tuple> read2 = txAcc.getAsync(makeKey(2));
+
+ txAcc.upsertAsync(makeValue(1, read1.join().doubleValue("balance") - DELTA));
+ txAcc.upsertAsync(makeValue(2, read2.join().doubleValue("balance") + DELTA));
+
+ tx.commit(); // Not necessary to wait for async ops expicitly before the commit.
+ });
+
+ Mockito.verify(tx).commit();
+
+ assertEquals(BALANCE_1 - DELTA, accounts.get(makeKey(1)).doubleValue("balance"));
+ assertEquals(BALANCE_2 + DELTA, accounts.get(makeKey(2)).doubleValue("balance"));
+ }
+
+ /**
+ * Tests a synchronous transaction over key-value view.
+ */
+ @Test
+ public void testTxSyncKeyValue() {
+ igniteTransactions.runInTransaction(tx -> {
+ KeyValueBinaryView txAcc = accounts.kvView().withTransaction(tx);
+
+ CompletableFuture<Tuple> read1 = txAcc.getAsync(makeKey(1));
+ CompletableFuture<Tuple> read2 = txAcc.getAsync(makeKey(2));
+
+ txAcc.putAsync(makeKey(1), makeValue(1, read1.join().doubleValue("balance") - DELTA));
+ txAcc.putAsync(makeKey(2), makeValue(2, read2.join().doubleValue("balance") + DELTA));
+
+ tx.commit(); // Not necessary to wait for async ops expicitly before the commit.
+ });
+
+ Mockito.verify(tx).commit();
+
+ assertEquals(BALANCE_1 - DELTA, accounts.get(makeKey(1)).doubleValue("balance"));
+ assertEquals(BALANCE_2 + DELTA, accounts.get(makeKey(2)).doubleValue("balance"));
+ }
+
+ /**
+ * Tests an asynchronous transaction.
+ */
+ @Test
+ public void testTxAsync() {
+ igniteTransactions.beginAsync().thenApply(tx -> accounts.withTransaction(tx)).
+ thenCompose(txAcc -> txAcc.getAsync(makeKey(1))
+ .thenCombine(txAcc.getAsync(makeKey(2)), (v1, v2) -> new Pair<>(v1, v2))
+ .thenCompose(pair -> allOf(
+ txAcc.upsertAsync(makeValue(1, pair.getFirst().doubleValue("balance") - DELTA)),
+ txAcc.upsertAsync(makeValue(2, pair.getSecond().doubleValue("balance") + DELTA))
+ )
+ )
+ .thenApply(ignore -> txAcc.transaction())
+ ).thenCompose(Transaction::commitAsync);
+
+ Mockito.verify(tx).commitAsync();
+
+ assertEquals(BALANCE_1 - DELTA, accounts.get(makeKey(1)).doubleValue("balance"));
+ assertEquals(BALANCE_2 + DELTA, accounts.get(makeKey(2)).doubleValue("balance"));
+ }
+
+ /**
+ * Tests an asynchronous transaction over key-value view.
+ */
+ @Test
+ public void testTxAsyncKeyValue() {
+ igniteTransactions.beginAsync().thenApply(tx -> accounts.kvView().withTransaction(tx)).
+ thenCompose(txAcc -> txAcc.getAsync(makeKey(1))
+ .thenCombine(txAcc.getAsync(makeKey(2)), (v1, v2) -> new Pair<>(v1, v2))
+ .thenCompose(pair -> allOf(
+ txAcc.putAsync(makeKey(1), makeValue(1, pair.getFirst().doubleValue("balance") - DELTA)),
+ txAcc.putAsync(makeKey(2), makeValue(2, pair.getSecond().doubleValue("balance") + DELTA))
+ )
+ )
+ .thenApply(ignore -> txAcc.transaction())
+ ).thenCompose(Transaction::commitAsync);
+
+ Mockito.verify(tx).commitAsync();
+
+ assertEquals(BALANCE_1 - DELTA, accounts.get(makeKey(1)).doubleValue("balance"));
+ assertEquals(BALANCE_2 + DELTA, accounts.get(makeKey(2)).doubleValue("balance"));
+ }
+
+ /**
+ * @param id The id.
+ * @return The key tuple.
+ */
+ private Tuple makeKey(long id) {
+ return accounts.tupleBuilder().set("accountNumber", id).build();
+ }
+
+ /**
+ * @param id The id.
+ * @param balance The balance.
+ * @return The value tuple.
+ */
+ private Tuple makeValue(long id, double balance) {
+ return accounts.tupleBuilder().set("accountNumber", id).set("balance", balance).build();
+ }
+}
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java b/modules/table/src/test/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
index 4f4bf29..25e5f55 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
@@ -29,6 +29,7 @@ import java.util.stream.Collectors;
import org.apache.ignite.internal.schema.BinaryRow;
import org.apache.ignite.internal.table.InternalTable;
import org.apache.ignite.schema.SchemaMode;
+import org.apache.ignite.tx.Transaction;
import org.jetbrains.annotations.NotNull;
/**
@@ -98,14 +99,14 @@ public class DummyInternalTableImpl implements InternalTable {
}
/** {@inheritDoc} */
- @Override public CompletableFuture<BinaryRow> get(@NotNull BinaryRow row) {
+ @Override public CompletableFuture<BinaryRow> get(@NotNull BinaryRow row, Transaction tx) {
assert row != null;
return CompletableFuture.completedFuture(store.get(extractAndWrapKey(row)));
}
/** {@inheritDoc} */
- @Override public CompletableFuture<Void> upsert(@NotNull BinaryRow row) {
+ @Override public CompletableFuture<Void> upsert(@NotNull BinaryRow row, Transaction tx) {
assert row != null;
store.put(extractAndWrapKey(row), row);
@@ -114,7 +115,8 @@ public class DummyInternalTableImpl implements InternalTable {
}
/** {@inheritDoc} */
- @Override public CompletableFuture<BinaryRow> getAndUpsert(@NotNull BinaryRow row) {
+ @Override public CompletableFuture<BinaryRow> getAndUpsert(@NotNull BinaryRow row,
+ Transaction tx) {
assert row != null;
final BinaryRow old = store.put(extractAndWrapKey(row), row);
@@ -123,7 +125,7 @@ public class DummyInternalTableImpl implements InternalTable {
}
/** {@inheritDoc} */
- @Override public @NotNull CompletableFuture<Boolean> delete(BinaryRow row) {
+ @Override public CompletableFuture<Boolean> delete(BinaryRow row, Transaction tx) {
assert row != null;
final KeyWrapper key = extractAndWrapKey(row);
@@ -133,7 +135,8 @@ public class DummyInternalTableImpl implements InternalTable {
}
/** {@inheritDoc} */
- @Override public @NotNull CompletableFuture<Collection<BinaryRow>> getAll(Collection<BinaryRow> keyRows) {
+ @Override public CompletableFuture<Collection<BinaryRow>> getAll(Collection<BinaryRow> keyRows,
+ Transaction tx) {
assert keyRows != null && !keyRows.isEmpty();
final List<BinaryRow> res = keyRows.stream()
@@ -145,7 +148,7 @@ public class DummyInternalTableImpl implements InternalTable {
}
/** {@inheritDoc} */
- @Override public @NotNull CompletableFuture<Void> upsertAll(Collection<BinaryRow> rows) {
+ @Override public CompletableFuture<Void> upsertAll(Collection<BinaryRow> rows, Transaction tx) {
assert rows != null && !rows.isEmpty();
rows.stream()
@@ -155,14 +158,14 @@ public class DummyInternalTableImpl implements InternalTable {
}
/** {@inheritDoc} */
- @Override public @NotNull CompletableFuture<Boolean> insert(BinaryRow row) {
+ @Override public CompletableFuture<Boolean> insert(BinaryRow row, Transaction tx) {
assert row != null;
return CompletableFuture.completedFuture(store.putIfAbsent(extractAndWrapKey(row), row) == null);
}
/** {@inheritDoc} */
- @Override public @NotNull CompletableFuture<Collection<BinaryRow>> insertAll(Collection<BinaryRow> rows) {
+ @Override public CompletableFuture<Collection<BinaryRow>> insertAll(Collection<BinaryRow> rows, Transaction tx) {
assert rows != null && !rows.isEmpty();
final List<BinaryRow> res = rows.stream()
@@ -174,7 +177,7 @@ public class DummyInternalTableImpl implements InternalTable {
}
/** {@inheritDoc} */
- @Override public @NotNull CompletableFuture<Boolean> replace(BinaryRow row) {
+ @Override public CompletableFuture<Boolean> replace(BinaryRow row, Transaction tx) {
assert row != null;
final KeyWrapper key = extractAndWrapKey(row);
@@ -187,7 +190,7 @@ public class DummyInternalTableImpl implements InternalTable {
}
/** {@inheritDoc} */
- @Override public @NotNull CompletableFuture<Boolean> replace(BinaryRow oldRow, BinaryRow newRow) {
+ @Override public CompletableFuture<Boolean> replace(BinaryRow oldRow, BinaryRow newRow, Transaction tx) {
assert oldRow != null;
assert newRow != null;
@@ -200,12 +203,12 @@ public class DummyInternalTableImpl implements InternalTable {
return CompletableFuture.completedFuture(equalValues(row, oldRow) && store.put(key, newRow) != null);
}
- @Override public @NotNull CompletableFuture<BinaryRow> getAndReplace(BinaryRow row) {
+ @Override public CompletableFuture<BinaryRow> getAndReplace(BinaryRow row, Transaction tx) {
return null;
}
/** {@inheritDoc} */
- @Override public @NotNull CompletableFuture<Boolean> deleteExact(BinaryRow row) {
+ @Override public CompletableFuture<Boolean> deleteExact(BinaryRow row, Transaction tx) {
assert row != null;
assert row.hasValue();
@@ -218,15 +221,16 @@ public class DummyInternalTableImpl implements InternalTable {
return CompletableFuture.completedFuture(equalValues(row, old) && store.remove(key) != null);
}
- @Override public @NotNull CompletableFuture<BinaryRow> getAndDelete(BinaryRow row) {
+ @Override public CompletableFuture<BinaryRow> getAndDelete(BinaryRow row, Transaction tx) {
return null;
}
- @Override public @NotNull CompletableFuture<Collection<BinaryRow>> deleteAll(Collection<BinaryRow> rows) {
+ @Override public CompletableFuture<Collection<BinaryRow>> deleteAll(Collection<BinaryRow> rows, Transaction tx) {
return null;
}
- @Override public @NotNull CompletableFuture<Collection<BinaryRow>> deleteAllExact(Collection<BinaryRow> rows) {
+ @Override public CompletableFuture<Collection<BinaryRow>> deleteAllExact(Collection<BinaryRow> rows,
+ Transaction tx) {
return null;
}