You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by vo...@apache.org on 2018/08/30 11:53:59 UTC
[37/38] ignite git commit: IGNITE-4191: MVCC and transactional SQL
support. Joint multi-man-years efforts of Semen Boikov, Igor Seliverstov,
Alexander Paschenko, Igor Sapego, Sergey Kalashnikov, Roman Kondakov,
Pavel Kuznetsov, Ivan Pavlukhin, Andrey Mas
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinTransactionsAbstractComplexSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinTransactionsAbstractComplexSelfTest.java b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinTransactionsAbstractComplexSelfTest.java
new file mode 100644
index 0000000..28c65a9
--- /dev/null
+++ b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinTransactionsAbstractComplexSelfTest.java
@@ -0,0 +1,1058 @@
+/*
+ * 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.jdbc.thin;
+
+import java.sql.BatchUpdateException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cache.query.SqlFieldsQuery;
+import org.apache.ignite.cache.query.annotations.QuerySqlField;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.IgniteInterruptedCheckedException;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.X;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.testframework.GridTestUtils;
+
+/**
+ * Test to check various transactional scenarios.
+ */
+public abstract class JdbcThinTransactionsAbstractComplexSelfTest extends JdbcThinAbstractSelfTest {
+ /** Client node index. */
+ final static int CLI_IDX = 1;
+
+ /**
+ * Closure to perform ordinary delete after repeatable read.
+ */
+ private final IgniteInClosure<Connection> afterReadDel = new IgniteInClosure<Connection>() {
+ @Override public void apply(Connection conn) {
+ execute(conn, "DELETE FROM \"Person\".Person where firstname = 'John'");
+ }
+ };
+
+ /**
+ * Closure to perform fast delete after repeatable read.
+ */
+ private final IgniteInClosure<Connection> afterReadFastDel = new IgniteInClosure<Connection>() {
+ @Override public void apply(Connection conn) {
+ execute(conn, "DELETE FROM \"Person\".Person where id = 1");
+ }
+ };
+
+ /**
+ * Closure to perform ordinary update after repeatable read.
+ */
+ private final IgniteInClosure<Connection> afterReadUpdate = new IgniteInClosure<Connection>() {
+ @Override public void apply(Connection conn) {
+ execute(conn, "UPDATE \"Person\".Person set firstname = 'Joe' where firstname = 'John'");
+ }
+ };
+
+ /**
+ * Closure to perform ordinary delete and rollback after repeatable read.
+ */
+ private final IgniteInClosure<Connection> afterReadDelAndRollback = new IgniteInClosure<Connection>() {
+ @Override public void apply(Connection conn) {
+ execute(conn, "DELETE FROM \"Person\".Person where firstname = 'John'");
+
+ rollback(conn);
+ }
+ };
+
+ /**
+ * Closure to perform fast delete after repeatable read.
+ */
+ private final IgniteInClosure<Connection> afterReadFastDelAndRollback = new IgniteInClosure<Connection>() {
+ @Override public void apply(Connection conn) {
+ execute(conn, "DELETE FROM \"Person\".Person where id = 1");
+
+ rollback(conn);
+ }
+ };
+
+ /**
+ * Closure to perform ordinary update and rollback after repeatable read.
+ */
+ private final IgniteInClosure<Connection> afterReadUpdateAndRollback = new IgniteInClosure<Connection>() {
+ @Override public void apply(Connection conn) {
+ execute(conn, "UPDATE \"Person\".Person set firstname = 'Joe' where firstname = 'John'");
+
+ rollback(conn);
+ }
+ };
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String testIgniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(testIgniteInstanceName);
+
+ cfg.setMvccEnabled(true);
+
+ CacheConfiguration<Integer, Person> ccfg = new CacheConfiguration<>("Person");
+
+ ccfg.setIndexedTypes(Integer.class, Person.class);
+
+ ccfg.getQueryEntities().iterator().next().setKeyFieldName("id");
+
+ ccfg.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL);
+
+ ccfg.setCacheMode(CacheMode.PARTITIONED);
+
+ cfg.setCacheConfiguration(ccfg);
+
+ // Let the node with index 1 be client node.
+ cfg.setClientMode(F.eq(testIgniteInstanceName, getTestIgniteInstanceName(CLI_IDX)));
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ super.beforeTest();
+
+ execute("ALTER TABLE \"Person\".person add if not exists cityid int");
+
+ execute("ALTER TABLE \"Person\".person add if not exists companyid int");
+
+ execute("CREATE TABLE City (id int primary key, name varchar, population int) WITH " +
+ "\"atomicity=transactional,template=partitioned,backups=3,cache_name=City\"");
+
+ execute("CREATE TABLE Company (id int, \"cityid\" int, name varchar, primary key (id, \"cityid\")) WITH " +
+ "\"atomicity=transactional,template=partitioned,backups=1,wrap_value=false,affinity_key=cityid," +
+ "cache_name=Company\"");
+
+ execute("CREATE TABLE Product (id int primary key, name varchar, companyid int) WITH " +
+ "\"atomicity=transactional,template=partitioned,backups=2,cache_name=Product\"");
+
+ execute("CREATE INDEX IF NOT EXISTS prodidx ON Product(companyid)");
+
+ execute("CREATE INDEX IF NOT EXISTS persidx ON \"Person\".person(cityid)");
+
+ insertPerson(1, "John", "Smith", 1, 1);
+
+ insertPerson(2, "Mike", "Johns", 1, 2);
+
+ insertPerson(3, "Sam", "Jules", 2, 2);
+
+ insertPerson(4, "Alex", "Pope", 2, 3);
+
+ insertPerson(5, "Peter", "Williams", 2, 3);
+
+ insertCity(1, "Los Angeles", 5000);
+
+ insertCity(2, "Seattle", 1500);
+
+ insertCity(3, "New York", 12000);
+
+ insertCity(4, "Cupertino", 400);
+
+ insertCompany(1, "Microsoft", 2);
+
+ insertCompany(2, "Google", 3);
+
+ insertCompany(3, "Facebook", 1);
+
+ insertCompany(4, "Uber", 1);
+
+ insertCompany(5, "Apple", 4);
+
+ insertProduct(1, "Search", 2);
+
+ insertProduct(2, "Windows", 1);
+
+ insertProduct(3, "Mac", 5);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ startGrid(0);
+
+ startGrid(1);
+
+ startGrid(2);
+
+ startGrid(3);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ stopAllGrids();
+
+ super.afterTestsStopped();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ execute("DELETE FROM \"Person\".Person");
+
+ execute("DROP TABLE City");
+
+ execute("DROP TABLE Company");
+
+ execute("DROP TABLE Product");
+
+ super.afterTest();
+ }
+
+ /**
+ *
+ */
+ public void testSingleDmlStatement() throws SQLException {
+ insertPerson(6, "John", "Doe", 2, 2);
+
+ assertEquals(Collections.singletonList(l(6, "John", "Doe", 2, 2)),
+ execute("SELECT * FROM \"Person\".Person where id = 6"));
+ }
+
+ /**
+ *
+ */
+ public void testMultipleDmlStatements() throws SQLException {
+ executeInTransaction(new TransactionClosure() {
+ @Override public void apply(Connection conn) {
+ insertPerson(conn, 6, "John", "Doe", 2, 2);
+
+ // https://issues.apache.org/jira/browse/IGNITE-6938 - we can only see results of
+ // UPDATE of what we have not inserted ourselves.
+ execute(conn, "UPDATE \"Person\".person SET lastname = 'Jameson' where lastname = 'Jules'");
+
+ execute(conn, "DELETE FROM \"Person\".person where id = 5");
+ }
+ });
+
+ assertEquals(l(
+ l(3, "Sam", "Jameson", 2, 2),
+ l(6, "John", "Doe", 2, 2)
+ ), execute("SELECT * FROM \"Person\".Person where id = 3 or id >= 5 order by id"));
+ }
+
+ /**
+ *
+ */
+ public void testBatchDmlStatements() throws SQLException {
+ doBatchedInsert();
+
+ assertEquals(l(
+ l(6, "John", "Doe", 2, 2),
+ l(7, "Mary", "Lee", 1, 3)
+ ), execute("SELECT * FROM \"Person\".Person where id > 5 order by id"));
+ }
+
+ /**
+ *
+ */
+ @SuppressWarnings("ThrowableResultOfMethodCallIgnored")
+ public void testBatchDmlStatementsIntermediateFailure() throws SQLException {
+ insertPerson(6, "John", "Doe", 2, 2);
+
+ IgniteException e = (IgniteException)GridTestUtils.assertThrows(null, new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ doBatchedInsert();
+
+ return null;
+ }
+ }, IgniteException.class, "Duplicate key during INSERT [key=KeyCacheObjectImpl " +
+ "[part=6, val=6, hasValBytes=true]]");
+
+ assertTrue(e.getCause() instanceof BatchUpdateException);
+
+ assertTrue(e.getCause().getMessage().contains("Duplicate key during INSERT [key=KeyCacheObjectImpl " +
+ "[part=6, val=6, hasValBytes=true]]"));
+
+ // First we insert id 7, then 6. Still, 7 is not in the cache as long as the whole batch has failed inside tx.
+ assertEquals(Collections.emptyList(), execute("SELECT * FROM \"Person\".Person where id > 6 order by id"));
+ }
+
+ /**
+ *
+ */
+ private void doBatchedInsert() throws SQLException {
+ executeInTransaction(new TransactionClosure() {
+ @Override public void apply(Connection conn) {
+ try {
+ try (PreparedStatement ps = conn.prepareStatement("INSERT INTO \"Person\".person " +
+ "(id, firstName, lastName, cityId, companyId) values (?, ?, ?, ?, ?)")) {
+ ps.setInt(1, 7);
+
+ ps.setString(2, "Mary");
+
+ ps.setString(3, "Lee");
+
+ ps.setInt(4, 1);
+
+ ps.setInt(5, 3);
+
+ ps.addBatch();
+
+ ps.setInt(1, 6);
+
+ ps.setString(2, "John");
+
+ ps.setString(3, "Doe");
+
+ ps.setInt(4, 2);
+
+ ps.setInt(5, 2);
+
+ ps.addBatch();
+
+ ps.executeBatch();
+ }
+ }
+ catch (SQLException e) {
+ throw new IgniteException(e);
+ }
+ }
+ });
+ }
+
+ /**
+ *
+ */
+ public void testInsertAndQueryMultipleCaches() throws SQLException {
+ executeInTransaction(new TransactionClosure() {
+ @Override public void apply(Connection conn) {
+ insertCity(conn, 5, "St Petersburg", 6000);
+
+ insertCompany(conn, 6, "VK", 5);
+
+ insertPerson(conn, 6, "Peter", "Sergeev", 5, 6);
+ }
+ });
+
+ try (Connection c = connect("distributedJoins=true")) {
+ assertEquals(l(l(5, "St Petersburg", 6000, 6, 5, "VK", 6, "Peter", "Sergeev", 5, 6)),
+ execute(c, "SELECT * FROM City left join Company on City.id = Company.\"cityid\" " +
+ "left join \"Person\".Person p on City.id = p.cityid WHERE p.id = 6 or company.id = 6"));
+ }
+ }
+
+ /**
+ *
+ */
+ public void testColocatedJoinSelectAndInsertInTransaction() throws SQLException {
+ // We'd like to put some Google into cities with over 1K population which don't have it yet
+ executeInTransaction(new TransactionClosure() {
+ @Override public void apply(Connection conn) {
+ List<Integer> ids = flat(execute(conn, "SELECT distinct City.id from City left join Company c on " +
+ "City.id = c.\"cityid\" where population >= 1000 and c.name <> 'Google' order by City.id"));
+
+ assertEqualsCollections(l(1, 2), ids);
+
+ int i = 5;
+
+ for (int l : ids)
+ insertCompany(conn, ++i, "Google", l);
+ }
+ });
+
+ assertEqualsCollections(l("Los Angeles", "Seattle", "New York"), flat(execute("SELECT City.name from City " +
+ "left join Company c on city.id = c.\"cityid\" WHERE c.name = 'Google' order by City.id")));
+ }
+
+ /**
+ *
+ */
+ public void testDistributedJoinSelectAndInsertInTransaction() throws SQLException {
+ try (Connection c = connect("distributedJoins=true")) {
+ // We'd like to put some Google into cities with over 1K population which don't have it yet
+ executeInTransaction(c, new TransactionClosure() {
+ @Override public void apply(Connection conn) {
+ List<?> res = flat(execute(conn, "SELECT p.id,p.name,c.id from Company c left join Product p on " +
+ "c.id = p.companyid left join City on city.id = c.\"cityid\" WHERE c.name <> 'Microsoft' " +
+ "and population < 1000"));
+
+ assertEqualsCollections(l(3, "Mac", 5), res);
+
+ insertProduct(conn, 4, (String)res.get(1), 1);
+ }
+ });
+ }
+
+ try (Connection c = connect("distributedJoins=true")) {
+ assertEqualsCollections(l("Windows", "Mac"), flat(execute(c, "SELECT p.name from Company c left join " +
+ "Product p on c.id = p.companyid WHERE c.name = 'Microsoft' order by p.id")));
+ }
+ }
+
+ /**
+ *
+ */
+ public void testInsertFromExpression() throws SQLException {
+ executeInTransaction(new TransactionClosure() {
+ @Override public void apply(Connection conn) {
+ execute(conn, "insert into city (id, name, population) values (? + 1, ?, ?)",
+ 8, "Moscow", 15000);
+ }
+ });
+ }
+
+ /**
+ *
+ */
+ public void testAutoRollback() throws SQLException {
+ try (Connection c = connect()) {
+ begin(c);
+
+ insertPerson(c, 6, "John", "Doe", 2, 2);
+ }
+
+ // Connection has not hung on close and update has not been applied.
+ assertTrue(personCache().query(new SqlFieldsQuery("SELECT * FROM \"Person\".Person WHERE id = 6"))
+ .getAll().isEmpty());
+ }
+
+ /**
+ *
+ */
+ public void testRepeatableReadWithConcurrentDelete() throws Exception {
+ doTestRepeatableRead(new IgniteInClosure<Connection>() {
+ @Override public void apply(Connection conn) {
+ execute(conn, "DELETE FROM \"Person\".Person where firstname = 'John'");
+ }
+ }, null);
+ }
+
+ /**
+ *
+ */
+ public void testRepeatableReadWithConcurrentFastDelete() throws Exception {
+ doTestRepeatableRead(new IgniteInClosure<Connection>() {
+ @Override public void apply(Connection conn) {
+ execute(conn, "DELETE FROM \"Person\".Person where id = 1");
+ }
+ }, null);
+ }
+
+ /**
+ *
+ */
+ public void testRepeatableReadWithConcurrentCacheRemove() throws Exception {
+ doTestRepeatableRead(new IgniteInClosure<Connection>() {
+ @Override public void apply(Connection conn) {
+ personCache().remove(1);
+ }
+ }, null);
+ }
+
+ /**
+ *
+ */
+ public void testRepeatableReadAndDeleteWithConcurrentDelete() throws Exception {
+ doTestRepeatableRead(new IgniteInClosure<Connection>() {
+ @Override public void apply(Connection conn) {
+ execute(conn, "DELETE FROM \"Person\".Person where firstname = 'John'");
+ }
+ }, afterReadDel);
+ }
+
+ /**
+ *
+ */
+ public void testRepeatableReadAndDeleteWithConcurrentFastDelete() throws Exception {
+ doTestRepeatableRead(new IgniteInClosure<Connection>() {
+ @Override public void apply(Connection conn) {
+ execute(conn, "DELETE FROM \"Person\".Person where id = 1");
+ }
+ }, afterReadDel);
+ }
+
+ /**
+ *
+ */
+ public void testRepeatableReadAndDeleteWithConcurrentCacheRemove() throws Exception {
+ doTestRepeatableRead(new IgniteInClosure<Connection>() {
+ @Override public void apply(Connection conn) {
+ personCache().remove(1);
+ }
+ }, afterReadDel);
+ }
+
+ /**
+ *
+ */
+ public void testRepeatableReadAndFastDeleteWithConcurrentDelete() throws Exception {
+ doTestRepeatableRead(new IgniteInClosure<Connection>() {
+ @Override public void apply(Connection conn) {
+ execute(conn, "DELETE FROM \"Person\".Person where firstname = 'John'");
+ }
+ }, afterReadFastDel);
+ }
+
+ /**
+ *
+ */
+ public void testRepeatableReadAndFastDeleteWithConcurrentFastDelete() throws Exception {
+ doTestRepeatableRead(new IgniteInClosure<Connection>() {
+ @Override public void apply(Connection conn) {
+ execute(conn, "DELETE FROM \"Person\".Person where id = 1");
+ }
+ }, afterReadFastDel);
+ }
+
+ /**
+ *
+ */
+ public void testRepeatableReadAndFastDeleteWithConcurrentCacheRemove() throws Exception {
+ doTestRepeatableRead(new IgniteInClosure<Connection>() {
+ @Override public void apply(Connection conn) {
+ personCache().remove(1);
+ }
+ }, afterReadFastDel);
+ }
+
+ /**
+ *
+ */
+ public void testRepeatableReadAndDeleteWithConcurrentDeleteAndRollback() throws Exception {
+ doTestRepeatableRead(new IgniteInClosure<Connection>() {
+ @Override public void apply(Connection conn) {
+ execute(conn, "DELETE FROM \"Person\".Person where firstname = 'John'");
+ }
+ }, afterReadDelAndRollback);
+ }
+
+ /**
+ *
+ */
+ public void testRepeatableReadAndDeleteWithConcurrentFastDeleteAndRollback() throws Exception {
+ doTestRepeatableRead(new IgniteInClosure<Connection>() {
+ @Override public void apply(Connection conn) {
+ execute(conn, "DELETE FROM \"Person\".Person where id = 1");
+ }
+ }, afterReadDelAndRollback);
+ }
+
+ /**
+ *
+ */
+ public void testRepeatableReadAndDeleteWithConcurrentCacheRemoveAndRollback() throws Exception {
+ doTestRepeatableRead(new IgniteInClosure<Connection>() {
+ @Override public void apply(Connection conn) {
+ personCache().remove(1);
+ }
+ }, afterReadDelAndRollback);
+ }
+
+ /**
+ *
+ */
+ public void testRepeatableReadAndFastDeleteWithConcurrentDeleteAndRollback() throws Exception {
+ doTestRepeatableRead(new IgniteInClosure<Connection>() {
+ @Override public void apply(Connection conn) {
+ execute(conn, "DELETE FROM \"Person\".Person where firstname = 'John'");
+ }
+ }, afterReadFastDelAndRollback);
+ }
+
+ /**
+ *
+ */
+ public void testRepeatableReadAndFastDeleteWithConcurrentFastDeleteAndRollback() throws Exception {
+ doTestRepeatableRead(new IgniteInClosure<Connection>() {
+ @Override public void apply(Connection conn) {
+ execute(conn, "DELETE FROM \"Person\".Person where id = 1");
+ }
+ }, afterReadFastDelAndRollback);
+ }
+
+ /**
+ *
+ */
+ public void testRepeatableReadAndFastDeleteWithConcurrentCacheRemoveAndRollback() throws Exception {
+ doTestRepeatableRead(new IgniteInClosure<Connection>() {
+ @Override public void apply(Connection conn) {
+ personCache().remove(1);
+ }
+ }, afterReadFastDelAndRollback);
+ }
+
+ /**
+ *
+ */
+ public void testRepeatableReadWithConcurrentUpdate() throws Exception {
+ doTestRepeatableRead(new IgniteInClosure<Connection>() {
+ @Override public void apply(Connection conn) {
+ execute(conn, "UPDATE \"Person\".Person SET lastname = 'Fix' where firstname = 'John'");
+ }
+ }, null);
+ }
+
+ /**
+ *
+ */
+ public void testRepeatableReadWithConcurrentCacheReplace() throws Exception {
+ doTestRepeatableRead(new IgniteInClosure<Connection>() {
+ @Override public void apply(Connection conn) {
+ Person p = new Person();
+
+ p.id = 1;
+ p.firstName = "Luke";
+ p.lastName = "Maxwell";
+
+ personCache().replace(1, p);
+ }
+ }, null);
+ }
+
+ /**
+ *
+ */
+ public void testRepeatableReadAndUpdateWithConcurrentUpdate() throws Exception {
+ doTestRepeatableRead(new IgniteInClosure<Connection>() {
+ @Override public void apply(Connection conn) {
+ execute(conn, "UPDATE \"Person\".Person SET lastname = 'Fix' where firstname = 'John'");
+ }
+ }, afterReadUpdate);
+ }
+
+ /**
+ *
+ */
+ public void testRepeatableReadAndUpdateWithConcurrentCacheReplace() throws Exception {
+ doTestRepeatableRead(new IgniteInClosure<Connection>() {
+ @Override public void apply(Connection conn) {
+ Person p = new Person();
+
+ p.id = 1;
+ p.firstName = "Luke";
+ p.lastName = "Maxwell";
+
+ personCache().replace(1, p);
+ }
+ }, afterReadUpdate);
+ }
+
+ /**
+ *
+ */
+ public void testRepeatableReadAndUpdateWithConcurrentUpdateAndRollback() throws Exception {
+ doTestRepeatableRead(new IgniteInClosure<Connection>() {
+ @Override public void apply(Connection conn) {
+ execute(conn, "UPDATE \"Person\".Person SET lastname = 'Fix' where firstname = 'John'");
+ }
+ }, afterReadUpdateAndRollback);
+ }
+
+ /**
+ *
+ */
+ public void testRepeatableReadAndUpdateWithConcurrentCacheReplaceAndRollback() throws Exception {
+ doTestRepeatableRead(new IgniteInClosure<Connection>() {
+ @Override public void apply(Connection conn) {
+ Person p = new Person();
+
+ p.id = 1;
+ p.firstName = "Luke";
+ p.lastName = "Maxwell";
+
+ personCache().replace(1, p);
+ }
+ }, afterReadUpdateAndRollback);
+ }
+
+ /**
+ * Perform repeatable reads and concurrent changes.
+ * @param concurrentWriteClo Updating closure.
+ * @param afterReadClo Closure making write changes that should also be made inside repeatable read transaction
+ * (must yield an exception).
+ * @throws Exception if failed.
+ */
+ @SuppressWarnings("ThrowableResultOfMethodCallIgnored")
+ private void doTestRepeatableRead(final IgniteInClosure<Connection> concurrentWriteClo,
+ final IgniteInClosure<Connection> afterReadClo) throws Exception {
+ final CountDownLatch repeatableReadLatch = new CountDownLatch(1);
+
+ final CountDownLatch initLatch = new CountDownLatch(1);
+
+ final IgniteInternalFuture<?> readFut = multithreadedAsync(new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ executeInTransaction(new TransactionClosure() {
+ @Override public void apply(Connection conn) {
+ List<?> before = flat(execute(conn, "SELECT * from \"Person\".Person where id = 1"));
+
+ assertEqualsCollections(l(1, "John", "Smith", 1, 1), before);
+
+ initLatch.countDown();
+
+ try {
+ U.await(repeatableReadLatch);
+ }
+ catch (IgniteInterruptedCheckedException e) {
+ throw new IgniteException(e);
+ }
+
+ List<?> after = flat(execute(conn, "SELECT * from \"Person\".Person where id = 1"));
+
+ assertEqualsCollections(before, after);
+
+ if (afterReadClo != null)
+ afterReadClo.apply(conn);
+ }
+ });
+
+ return null;
+ }
+ }, 1);
+
+ IgniteInternalFuture<?> conModFut = multithreadedAsync(new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ executeInTransaction(new TransactionClosure() {
+ @Override public void apply(Connection conn) {
+ try {
+ U.await(initLatch);
+ }
+ catch (IgniteInterruptedCheckedException e) {
+ throw new IgniteException(e);
+ }
+
+ concurrentWriteClo.apply(conn);
+
+ repeatableReadLatch.countDown();
+ }
+ });
+
+ return null;
+ }
+ }, 1);
+
+ conModFut.get();
+
+ if (afterReadClo != null) {
+ IgniteCheckedException ex = (IgniteCheckedException)GridTestUtils.assertThrows(null, new Callable() {
+ @Override public Object call() throws Exception {
+ readFut.get();
+
+ return null;
+ }
+ }, IgniteCheckedException.class, "Mvcc version mismatch.");
+
+ assertTrue(X.hasCause(ex, SQLException.class));
+
+ assertTrue(X.getCause(ex).getMessage().contains("Mvcc version mismatch."));
+ }
+ else
+ readFut.get();
+ }
+
+ /**
+ * Create a new connection, a new transaction and run given closure in its scope.
+ * @param clo Closure.
+ * @throws SQLException if failed.
+ */
+ private void executeInTransaction(TransactionClosure clo) throws SQLException {
+ try (Connection conn = connect()) {
+ executeInTransaction(conn, clo);
+ }
+ }
+
+ /**
+ * Create a new transaction and run given closure in its scope.
+ * @param conn Connection.
+ * @param clo Closure.
+ * @throws SQLException if failed.
+ */
+ private void executeInTransaction(Connection conn, TransactionClosure clo) throws SQLException {
+ begin(conn);
+
+ clo.apply(conn);
+
+ commit(conn);
+ }
+
+ /**
+ * @return Auto commit strategy for this test.
+ */
+ abstract boolean autoCommit();
+
+ /**
+ * @param c Connection to begin a transaction on.
+ */
+ private void begin(Connection c) throws SQLException {
+ if (autoCommit())
+ execute(c, "BEGIN");
+ }
+
+ /**
+ * @param c Connection to begin a transaction on.
+ */
+ private void commit(Connection c) throws SQLException {
+ if (autoCommit())
+ execute(c, "COMMIT");
+ else
+ c.commit();
+ }
+
+ /**
+ * @param c Connection to rollback a transaction on.
+ */
+ private void rollback(Connection c) {
+ try {
+ if (autoCommit())
+ execute(c, "ROLLBACK");
+ else
+ c.rollback();
+ }
+ catch (SQLException e) {
+ throw new IgniteException(e);
+ }
+ }
+
+ /**
+ * @param sql Statement.
+ * @param args Arguments.
+ * @return Result set.
+ * @throws SQLException if failed.
+ */
+ List<List<?>> execute(String sql, Object... args) throws SQLException {
+ try (Connection c = connect()) {
+ c.setAutoCommit(true);
+
+ return execute(c, sql, args);
+ }
+ }
+
+ /**
+ * @param sql Statement.
+ * @param args Arguments.
+ * @return Result set.
+ * @throws RuntimeException if failed.
+ */
+ protected List<List<?>> execute(Connection conn, String sql, Object... args) {
+ try {
+ return super.execute(conn, sql, args);
+ }
+ catch (SQLException e) {
+ throw new IgniteException(e);
+ }
+ }
+
+ /**
+ * @return New connection to default node.
+ * @throws SQLException if failed.
+ */
+ private Connection connect() throws SQLException {
+ return connect(null);
+ }
+
+ /**
+ * @param params Connection parameters.
+ * @return New connection to default node.
+ * @throws SQLException if failed.
+ */
+ private Connection connect(String params) throws SQLException {
+ Connection c = connect(node(), params);
+
+ c.setAutoCommit(false);
+
+ return c;
+ }
+
+ /**
+ * @param node Node to connect to.
+ * @param params Connection parameters.
+ * @return Thin JDBC connection to specified node.
+ */
+ protected Connection connect(IgniteEx node, String params) {
+ try {
+ return super.connect(node, params);
+ }
+ catch (SQLException e) {
+ throw new AssertionError(e);
+ }
+ }
+
+ /**
+ * @return Default node to fire queries from.
+ */
+ private IgniteEx node() {
+ return grid(nodeIndex());
+ }
+
+ /**
+ * @return {@link Person} cache.
+ */
+ private IgniteCache<Integer, Person> personCache() {
+ return node().cache("Person");
+ }
+
+ /**
+ * @return Node index to fire queries from.
+ */
+ abstract int nodeIndex();
+
+ /**
+ * @param id New person's id.
+ * @param firstName First name.
+ * @param lastName Second name.
+ * @param cityId City id.
+ * @param companyId Company id.
+ * @throws SQLException if failed.
+ */
+ private void insertPerson(final int id, final String firstName, final String lastName, final int cityId,
+ final int companyId) throws SQLException {
+ executeInTransaction(new TransactionClosure() {
+ @Override public void apply(Connection conn) {
+ insertPerson(conn, id, firstName, lastName, cityId, companyId);
+ }
+ });
+ }
+
+ /**
+ * @param c Connection.
+ * @param id New person's id.
+ * @param firstName First name.
+ * @param lastName Second name.
+ * @param cityId City id.
+ * @param companyId Company id.
+ */
+ private void insertPerson(Connection c, int id, String firstName, String lastName, int cityId, int companyId) {
+ execute(c, "INSERT INTO \"Person\".person (id, firstName, lastName, cityId, companyId) values (?, ?, ?, ?, ?)",
+ id, firstName, lastName, cityId, companyId);
+ }
+
+ /**
+ * @param id New city's id.
+ * @param name City name.
+ * @param population Number of people.
+ * @throws SQLException if failed.
+ */
+ private void insertCity(final int id, final String name, final int population) throws SQLException {
+ executeInTransaction(new TransactionClosure() {
+ @Override public void apply(Connection conn) {
+ insertCity(conn, id, name, population);
+ }
+ });
+ }
+
+ /**
+ * @param c Connection.
+ * @param id New city's id.
+ * @param name City name.
+ * @param population Number of people.
+ */
+ private void insertCity(Connection c, int id, String name, int population) {
+ execute(c, "INSERT INTO city (id, name, population) values (?, ?, ?)", id, name, population);
+ }
+
+ /**
+ * @param id New company's id.
+ * @param name Company name.
+ * @param cityId City id.
+ * @throws SQLException if failed.
+ */
+ private void insertCompany(final int id, final String name, final int cityId) throws SQLException {
+ executeInTransaction(new TransactionClosure() {
+ @Override public void apply(Connection conn) {
+ insertCompany(conn, id, name, cityId);
+ }
+ });
+ }
+
+ /**
+ * @param c Connection.
+ * @param id New company's id.
+ * @param name Company name.
+ * @param cityId City id.
+ */
+ private void insertCompany(Connection c, int id, String name, int cityId) {
+ execute(c, "INSERT INTO company (id, name, \"cityid\") values (?, ?, ?)", id, name, cityId);
+ }
+
+ /**
+ * @param id New product's id.
+ * @param name Product name.
+ * @param companyId Company id..
+ * @throws SQLException if failed.
+ */
+ private void insertProduct(final int id, final String name, final int companyId) throws SQLException {
+ executeInTransaction(new TransactionClosure() {
+ @Override public void apply(Connection conn) {
+ insertProduct(conn, id, name, companyId);
+ }
+ });
+ }
+
+ /**
+ * @param c Connection.
+ * @param id New product's id.
+ * @param name Product name.
+ * @param companyId Company id..
+ */
+ private void insertProduct(Connection c, int id, String name, int companyId) {
+ execute(c, "INSERT INTO product (id, name, companyid) values (?, ?, ?)", id, name, companyId);
+ }
+
+ /**
+ * Person class.
+ */
+ private final static class Person {
+ /** */
+ @QuerySqlField
+ public int id;
+
+ /** */
+ @QuerySqlField
+ public String firstName;
+
+ /** */
+ @QuerySqlField
+ public String lastName;
+ }
+
+ /**
+ * Closure to be executed in scope of a transaction.
+ */
+ private abstract class TransactionClosure implements IgniteInClosure<Connection> {
+ // No-op.
+ }
+
+ /**
+ * @return List of given arguments.
+ */
+ private static List<?> l(Object... args) {
+ return F.asList(args);
+ }
+
+ /**
+ * Flatten rows.
+ * @param rows Rows.
+ * @return Rows as a single list.
+ */
+ @SuppressWarnings("unchecked")
+ private static <T> List<T> flat(Collection<? extends Collection<?>> rows) {
+ return new ArrayList<>(F.flatCollections((Collection<? extends Collection<T>>)rows));
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinTransactionsClientAutoCommitComplexSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinTransactionsClientAutoCommitComplexSelfTest.java b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinTransactionsClientAutoCommitComplexSelfTest.java
new file mode 100644
index 0000000..d5b505a
--- /dev/null
+++ b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinTransactionsClientAutoCommitComplexSelfTest.java
@@ -0,0 +1,33 @@
+/*
+ * 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.jdbc.thin;
+
+/**
+ *
+ */
+public class JdbcThinTransactionsClientAutoCommitComplexSelfTest extends JdbcThinTransactionsAbstractComplexSelfTest {
+ /** {@inheritDoc} */
+ @Override boolean autoCommit() {
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override int nodeIndex() {
+ return CLI_IDX;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinTransactionsClientNoAutoCommitComplexSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinTransactionsClientNoAutoCommitComplexSelfTest.java b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinTransactionsClientNoAutoCommitComplexSelfTest.java
new file mode 100644
index 0000000..7fa69fd
--- /dev/null
+++ b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinTransactionsClientNoAutoCommitComplexSelfTest.java
@@ -0,0 +1,33 @@
+/*
+ * 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.jdbc.thin;
+
+/**
+ *
+ */
+public class JdbcThinTransactionsClientNoAutoCommitComplexSelfTest extends JdbcThinTransactionsAbstractComplexSelfTest {
+ /** {@inheritDoc} */
+ @Override boolean autoCommit() {
+ return false;
+ }
+
+ /** {@inheritDoc} */
+ @Override int nodeIndex() {
+ return CLI_IDX;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinTransactionsSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinTransactionsSelfTest.java b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinTransactionsSelfTest.java
new file mode 100644
index 0000000..1619996
--- /dev/null
+++ b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinTransactionsSelfTest.java
@@ -0,0 +1,447 @@
+/*
+ * 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.jdbc.thin;
+
+import java.sql.BatchUpdateException;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.concurrent.Callable;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cache.query.SqlFieldsQuery;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.binary.BinaryMarshaller;
+import org.apache.ignite.internal.processors.query.NestedTxMode;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.GridStringLogger;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Tests to check behavior with transactions on.
+ */
+public class JdbcThinTransactionsSelfTest extends JdbcThinAbstractSelfTest {
+ /** IP finder. */
+ private static final TcpDiscoveryIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true);
+
+ /** */
+ private static final String URL = "jdbc:ignite:thin://127.0.0.1";
+
+ /** Logger. */
+ private GridStringLogger log;
+
+ /** {@inheritDoc} */
+ @SuppressWarnings("deprecation")
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.setMvccEnabled(true);
+
+ cfg.setCacheConfiguration(cacheConfiguration(DEFAULT_CACHE_NAME));
+
+ TcpDiscoverySpi disco = new TcpDiscoverySpi();
+
+ disco.setIpFinder(IP_FINDER);
+
+ cfg.setDiscoverySpi(disco);
+
+ cfg.setMarshaller(new BinaryMarshaller());
+
+ cfg.setGridLogger(log = new GridStringLogger());
+
+ return cfg;
+ }
+
+ /**
+ * @param name Cache name.
+ * @return Cache configuration.
+ * @throws Exception In case of error.
+ */
+ private CacheConfiguration cacheConfiguration(@NotNull String name) throws Exception {
+ CacheConfiguration cfg = defaultCacheConfiguration();
+
+ cfg.setName(name);
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ super.beforeTest();
+
+ startGrid(0);
+
+ try (Connection c = c(true, NestedTxMode.ERROR)) {
+ try (Statement s = c.createStatement()) {
+ s.execute("CREATE TABLE INTS (k int primary key, v int) WITH \"cache_name=ints,wrap_value=false," +
+ "atomicity=transactional\"");
+ }
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ stopAllGrids();
+ }
+
+ /**
+ * @param autoCommit Auto commit mode.
+ * @param nestedTxMode Nested transactions mode.
+ * @return Connection.
+ * @throws SQLException if failed.
+ */
+ private static Connection c(boolean autoCommit, NestedTxMode nestedTxMode) throws SQLException {
+ Connection res = DriverManager.getConnection(URL + "/?nestedTransactionsMode=" + nestedTxMode.name());
+
+ res.setAutoCommit(autoCommit);
+
+ return res;
+ }
+
+ /**
+ *
+ */
+ public void testTransactionsBeginCommitRollback() throws IgniteCheckedException {
+ final AtomicBoolean stop = new AtomicBoolean();
+
+ IgniteInternalFuture<?> fut = GridTestUtils.runMultiThreadedAsync(new Runnable() {
+ @Override public void run() {
+ try {
+ try (Connection c = c(false, NestedTxMode.ERROR)) {
+ while (!stop.get()) {
+ try (Statement s = c.createStatement()) {
+ s.execute("BEGIN");
+
+ c.commit();
+
+ s.execute("BEGIN");
+
+ c.rollback();
+ }
+ }
+ }
+ }
+ catch (SQLException e) {
+ throw new AssertionError(e);
+ }
+ }
+ }, 8, "jdbc-transactions");
+
+ U.sleep(5000);
+
+ stop.set(true);
+
+ fut.get();
+ }
+
+ /**
+ *
+ */
+ public void testTransactionsBeginCommitRollbackAutocommit() throws IgniteCheckedException {
+ GridTestUtils.runMultiThreadedAsync(new Runnable() {
+ @Override public void run() {
+ try {
+ try (Connection c = c(true, NestedTxMode.ERROR)) {
+ try (Statement s = c.createStatement()) {
+ s.execute("BEGIN");
+
+ s.execute("COMMIT");
+
+ s.execute("BEGIN");
+
+ s.execute("ROLLBACK");
+ }
+ }
+ }
+ catch (SQLException e) {
+ throw new AssertionError(e);
+ }
+ }
+ }, 8, "jdbc-transactions").get();
+ }
+
+ /**
+ *
+ */
+ @SuppressWarnings("ThrowableResultOfMethodCallIgnored")
+ public void testIgnoreNestedTxAutocommitOff() throws SQLException {
+ try (Connection c = c(false, NestedTxMode.IGNORE)) {
+ doNestedTxStart(c, false);
+ }
+
+ assertTrue(log.toString().contains("ignoring BEGIN command"));
+ }
+
+ /**
+ *
+ */
+ @SuppressWarnings("ThrowableResultOfMethodCallIgnored")
+ public void testCommitNestedTxAutocommitOff() throws SQLException {
+ try (Connection c = c(false, NestedTxMode.COMMIT)) {
+ doNestedTxStart(c, false);
+ }
+
+ assertFalse(log.toString().contains("ignoring BEGIN command"));
+ }
+
+ /**
+ *
+ */
+ @SuppressWarnings("ThrowableResultOfMethodCallIgnored")
+ public void testErrorNestedTxAutocommitOff() throws SQLException {
+ GridTestUtils.assertThrows(null, new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ try (Connection c = c(false, NestedTxMode.ERROR)) {
+ doNestedTxStart(c, false);
+ }
+
+ throw new AssertionError();
+ }
+ }, SQLException.class, "Transaction has already been started.");
+ }
+
+ /**
+ *
+ */
+ @SuppressWarnings("ThrowableResultOfMethodCallIgnored")
+ public void testIgnoreNestedTxAutocommitOn() throws SQLException {
+ try (Connection c = c(true, NestedTxMode.IGNORE)) {
+ doNestedTxStart(c, false);
+ }
+
+ assertTrue(log.toString().contains("ignoring BEGIN command"));
+ }
+
+ /**
+ *
+ */
+ @SuppressWarnings("ThrowableResultOfMethodCallIgnored")
+ public void testCommitNestedTxAutocommitOn() throws SQLException {
+ try (Connection c = c(true, NestedTxMode.COMMIT)) {
+ doNestedTxStart(c, false);
+ }
+
+ assertFalse(log.toString().contains("ignoring BEGIN command"));
+ }
+
+ /**
+ *
+ */
+ @SuppressWarnings("ThrowableResultOfMethodCallIgnored")
+ public void testErrorNestedTxAutocommitOn() throws SQLException {
+ GridTestUtils.assertThrows(null, new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ try (Connection c = c(true, NestedTxMode.ERROR)) {
+ doNestedTxStart(c, false);
+ }
+
+ throw new AssertionError();
+ }
+ }, SQLException.class, "Transaction has already been started.");
+ }
+
+ /**
+ *
+ */
+ @SuppressWarnings("ThrowableResultOfMethodCallIgnored")
+ public void testIgnoreNestedTxAutocommitOffBatched() throws SQLException {
+ try (Connection c = c(false, NestedTxMode.IGNORE)) {
+ doNestedTxStart(c, true);
+ }
+
+ assertTrue(log.toString().contains("ignoring BEGIN command"));
+ }
+
+ /**
+ *
+ */
+ @SuppressWarnings("ThrowableResultOfMethodCallIgnored")
+ public void testCommitNestedTxAutocommitOffBatched() throws SQLException {
+ try (Connection c = c(false, NestedTxMode.COMMIT)) {
+ doNestedTxStart(c, true);
+ }
+
+ assertFalse(log.toString().contains("ignoring BEGIN command"));
+ }
+
+ /**
+ *
+ */
+ @SuppressWarnings("ThrowableResultOfMethodCallIgnored")
+ public void testErrorNestedTxAutocommitOffBatched() throws SQLException {
+ GridTestUtils.assertThrows(null, new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ try (Connection c = c(false, NestedTxMode.ERROR)) {
+ doNestedTxStart(c, true);
+ }
+
+ throw new AssertionError();
+ }
+ }, BatchUpdateException.class, "Transaction has already been started.");
+ }
+
+ /**
+ *
+ */
+ @SuppressWarnings("ThrowableResultOfMethodCallIgnored")
+ public void testIgnoreNestedTxAutocommitOnBatched() throws SQLException {
+ try (Connection c = c(true, NestedTxMode.IGNORE)) {
+ doNestedTxStart(c, true);
+ }
+
+ assertTrue(log.toString().contains("ignoring BEGIN command"));
+ }
+
+ /**
+ *
+ */
+ @SuppressWarnings("ThrowableResultOfMethodCallIgnored")
+ public void testCommitNestedTxAutocommitOnBatched() throws SQLException {
+ try (Connection c = c(true, NestedTxMode.COMMIT)) {
+ doNestedTxStart(c, true);
+ }
+
+ assertFalse(log.toString().contains("ignoring BEGIN command"));
+ }
+
+ /**
+ *
+ */
+ @SuppressWarnings("ThrowableResultOfMethodCallIgnored")
+ public void testErrorNestedTxAutocommitOnBatched() throws SQLException {
+ GridTestUtils.assertThrows(null, new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ try (Connection c = c(true, NestedTxMode.ERROR)) {
+ doNestedTxStart(c, true);
+ }
+
+ throw new AssertionError();
+ }
+ }, BatchUpdateException.class, "Transaction has already been started.");
+ }
+
+ /**
+ * Try to start nested transaction via batch as well as separate statements.
+ * @param conn Connection.
+ * @param batched Whether {@link Statement#executeBatch()} should be used.
+ * @throws SQLException if failed.
+ */
+ private void doNestedTxStart(Connection conn, boolean batched) throws SQLException {
+ try (Statement s = conn.createStatement()) {
+ s.executeQuery("SELECT * FROM INTS");
+
+ if (batched) {
+ s.addBatch("BEGIN");
+
+ s.addBatch("BEGIN");
+
+ s.executeBatch();
+ }
+ else {
+ s.execute("BEGIN");
+
+ s.execute("BEGIN");
+ }
+ }
+ }
+
+ /**
+ * @throws SQLException if failed.
+ */
+ public void testAutoCommitSingle() throws SQLException {
+ doTestAutoCommit(false);
+ }
+
+ /**
+ * @throws SQLException if failed.
+ */
+ public void testAutoCommitBatched() throws SQLException {
+ doTestAutoCommit(true);
+ }
+
+ /**
+ * @param batched Batch mode flag.
+ * @throws SQLException if failed.
+ */
+ private void doTestAutoCommit(boolean batched) throws SQLException {
+ IgniteCache<Integer, ?> cache = grid(0).cache("ints");
+
+ try (Connection c = c(false, NestedTxMode.ERROR)) {
+ try (Statement s = c.createStatement()) {
+ assertFalse(s.executeQuery("SELECT * from INTS").next());
+
+ if (batched) {
+ s.addBatch("INSERT INTO INTS(k, v) values(1, 1)");
+
+ s.executeBatch();
+ }
+ else
+ s.execute("INSERT INTO INTS(k, v) values(1, 1)");
+
+ // We haven't committed anything yet - this check shows that autoCommit flag is in effect.
+ assertTrue(cache.query(new SqlFieldsQuery("SELECT * from INTS")).getAll().isEmpty());
+
+ // We should see own updates.
+ assertTrue(s.executeQuery("SELECT * from INTS").next());
+
+ c.commit();
+
+ c.setAutoCommit(true);
+
+ assertEquals(1, cache.get(1));
+
+ assertTrue(s.executeQuery("SELECT * from INTS").next());
+ }
+ }
+ }
+
+ /**
+ * Test that exception in one of the statements does not kill connection worker altogether.
+ * @throws SQLException if failed.
+ */
+ @SuppressWarnings("ThrowableResultOfMethodCallIgnored")
+ public void testExceptionHandling() throws SQLException {
+ try (Connection c = c(true, NestedTxMode.ERROR)) {
+ try (Statement s = c.createStatement()) {
+ s.execute("INSERT INTO INTS(k, v) values(1, 1)");
+
+ assertEquals(1, grid(0).cache("ints").get(1));
+
+ GridTestUtils.assertThrows(null, new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ s.execute("INSERT INTO INTS(x, y) values(1, 1)");
+
+ return null;
+ }
+ }, SQLException.class, "Failed to parse query");
+
+ s.execute("INSERT INTO INTS(k, v) values(2, 2)");
+
+ assertEquals(2, grid(0).cache("ints").get(2));
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinTransactionsServerAutoCommitComplexSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinTransactionsServerAutoCommitComplexSelfTest.java b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinTransactionsServerAutoCommitComplexSelfTest.java
new file mode 100644
index 0000000..3c473ab
--- /dev/null
+++ b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinTransactionsServerAutoCommitComplexSelfTest.java
@@ -0,0 +1,33 @@
+/*
+ * 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.jdbc.thin;
+
+/**
+ *
+ */
+public class JdbcThinTransactionsServerAutoCommitComplexSelfTest extends JdbcThinTransactionsAbstractComplexSelfTest {
+ /** {@inheritDoc} */
+ @Override boolean autoCommit() {
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override int nodeIndex() {
+ return 0;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinTransactionsServerNoAutoCommitComplexSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinTransactionsServerNoAutoCommitComplexSelfTest.java b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinTransactionsServerNoAutoCommitComplexSelfTest.java
new file mode 100644
index 0000000..655d4c5
--- /dev/null
+++ b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinTransactionsServerNoAutoCommitComplexSelfTest.java
@@ -0,0 +1,33 @@
+/*
+ * 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.jdbc.thin;
+
+/**
+ *
+ */
+public class JdbcThinTransactionsServerNoAutoCommitComplexSelfTest extends JdbcThinTransactionsAbstractComplexSelfTest {
+ /** {@inheritDoc} */
+ @Override boolean autoCommit() {
+ return false;
+ }
+
+ /** {@inheritDoc} */
+ @Override int nodeIndex() {
+ return 0;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java
----------------------------------------------------------------------
diff --git a/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java b/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java
index 99cf849..51365f3 100644
--- a/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java
+++ b/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java
@@ -38,6 +38,8 @@ import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
import java.util.UUID;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxQueryEnlistRequest;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxQueryResultsEnlistRequest;
import org.apache.ignite.internal.util.IgniteUtils;
import org.apache.ignite.internal.GridCodegenConverter;
import org.apache.ignite.internal.GridDirectCollection;
@@ -168,7 +170,7 @@ public class MessageCodeGenerator {
// gen.generateAll(true);
-// gen.generateAndWrite(GridChangeGlobalStateMessageResponse.class);
+ gen.generateAndWrite(GridNearTxQueryResultsEnlistRequest.class);
// gen.generateAndWrite(GridNearAtomicUpdateRequest.class);
@@ -235,6 +237,8 @@ public class MessageCodeGenerator {
// gen.generateAndWrite(GridCacheVersionEx.class);
// gen.generateAndWrite(GridH2DmlRequest.class);
// gen.generateAndWrite(GridH2DmlResponse.class);
+// gen.generateAndWrite(GridNearTxEnlistRequest.class);
+// gen.generateAndWrite(GridNearTxEnlistResponse.class);
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/configuration/CacheConfiguration.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/configuration/CacheConfiguration.java b/modules/core/src/main/java/org/apache/ignite/configuration/CacheConfiguration.java
index 2e35f37..fb3789d 100644
--- a/modules/core/src/main/java/org/apache/ignite/configuration/CacheConfiguration.java
+++ b/modules/core/src/main/java/org/apache/ignite/configuration/CacheConfiguration.java
@@ -1019,6 +1019,7 @@ public class CacheConfiguration<K, V> extends MutableConfiguration<K, V> {
* @param atomicityMode Cache atomicity mode.
* @return {@code this} for chaining.
*/
+ @SuppressWarnings("unchecked")
public CacheConfiguration<K, V> setAtomicityMode(CacheAtomicityMode atomicityMode) {
this.atomicityMode = atomicityMode;
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/configuration/IgniteConfiguration.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/configuration/IgniteConfiguration.java b/modules/core/src/main/java/org/apache/ignite/configuration/IgniteConfiguration.java
index ab54709..3060caa 100644
--- a/modules/core/src/main/java/org/apache/ignite/configuration/IgniteConfiguration.java
+++ b/modules/core/src/main/java/org/apache/ignite/configuration/IgniteConfiguration.java
@@ -214,6 +214,13 @@ public class IgniteConfiguration {
/** Default timeout after which long query warning will be printed. */
public static final long DFLT_LONG_QRY_WARN_TIMEOUT = 3000;
+ /** Default size of MVCC vacuum thread pool. */
+ public static final int DFLT_MVCC_VACUUM_THREAD_CNT = 2;
+
+ /** Default time interval between vacuum process runs (ms). */
+ public static final int DFLT_MVCC_VACUUM_TIME_INTERVAL = 5000;
+
+
/** Optional local Ignite instance name. */
private String igniteInstanceName;
@@ -487,6 +494,15 @@ public class IgniteConfiguration {
/** Client connector configuration. */
private ClientConnectorConfiguration cliConnCfg = ClientListenerProcessor.DFLT_CLI_CFG;
+ /** Flag whether MVCC is enabled. */
+ private boolean mvccEnabled;
+
+ /** Size of MVCC vacuum thread pool. */
+ private int mvccVacuumThreadCnt = DFLT_MVCC_VACUUM_THREAD_CNT;
+
+ /** Time interval between vacuum process runs (ms). */
+ private int mvccVacuumTimeInterval = DFLT_MVCC_VACUUM_TIME_INTERVAL;
+
/** User authentication enabled. */
private boolean authEnabled;
@@ -535,6 +551,7 @@ public class IgniteConfiguration {
addrRslvr = cfg.getAddressResolver();
allResolversPassReq = cfg.isAllSegmentationResolversPassRequired();
atomicCfg = cfg.getAtomicConfiguration();
+ authEnabled = cfg.isAuthenticationEnabled();
autoActivation = cfg.isAutoActivationEnabled();
binaryCfg = cfg.getBinaryConfiguration();
dsCfg = cfg.getDataStorageConfiguration();
@@ -578,6 +595,9 @@ public class IgniteConfiguration {
metricsLogFreq = cfg.getMetricsLogFrequency();
metricsUpdateFreq = cfg.getMetricsUpdateFrequency();
mgmtPoolSize = cfg.getManagementThreadPoolSize();
+ mvccEnabled = cfg.isMvccEnabled();
+ mvccVacuumThreadCnt = cfg.mvccVacuumThreadCnt;
+ mvccVacuumTimeInterval = cfg.mvccVacuumTimeInterval;
netTimeout = cfg.getNetworkTimeout();
nodeId = cfg.getNodeId();
odbcCfg = cfg.getOdbcConfiguration();
@@ -612,7 +632,6 @@ public class IgniteConfiguration {
utilityCachePoolSize = cfg.getUtilityCacheThreadPoolSize();
waitForSegOnStart = cfg.isWaitForSegmentOnStart();
warmupClos = cfg.getWarmupClosure();
- authEnabled = cfg.isAuthenticationEnabled();
}
/**
@@ -2984,6 +3003,69 @@ public class IgniteConfiguration {
}
/**
+ * Whether or not MVCC is enabled.
+ *
+ * @return {@code True} if MVCC is enabled.
+ */
+ public boolean isMvccEnabled() {
+ return mvccEnabled;
+ }
+
+ /**
+ * Sets MVCC enabled flag.
+ *
+ * @param mvccEnabled MVCC enabled flag.
+ * @return {@code this} for chaining.
+ */
+ public IgniteConfiguration setMvccEnabled(boolean mvccEnabled) {
+ this.mvccEnabled = mvccEnabled;
+
+ return this;
+ }
+
+ /**
+ * Returns number of MVCC vacuum cleanup threads.
+ *
+ * @return Number of MVCC vacuum cleanup threads.
+ */
+ public int getMvccVacuumThreadCnt() {
+ return mvccVacuumThreadCnt;
+ }
+
+ /**
+ * Sets number of MVCC vacuum cleanup threads.
+ *
+ * @param mvccVacuumThreadCnt Number of MVCC vacuum cleanup threads.
+ * @return {@code this} for chaining.
+ */
+ public IgniteConfiguration setMvccVacuumThreadCnt(int mvccVacuumThreadCnt) {
+ this.mvccVacuumThreadCnt = mvccVacuumThreadCnt;
+
+ return this;
+ }
+
+ /**
+ * Returns time interval between vacuum runs.
+ *
+ * @return Time interval between vacuum runs.
+ */
+ public int getMvccVacuumTimeInterval() {
+ return mvccVacuumTimeInterval;
+ }
+
+ /**
+ * Sets time interval between vacuum runs.
+ *
+ * @param mvccVacuumTimeInterval Time interval between vacuum runs.
+ * @return {@code this} for chaining.
+ */
+ public IgniteConfiguration setMvccVacuumTimeInterval(int mvccVacuumTimeInterval) {
+ this.mvccVacuumTimeInterval = mvccVacuumTimeInterval;
+
+ return this;
+ }
+
+ /**
* Returns {@code true} if user authentication is enabled for cluster. Otherwise returns {@code false}.
* Default value is false; authentication is disabled.
*
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/GridComponent.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridComponent.java b/modules/core/src/main/java/org/apache/ignite/internal/GridComponent.java
index 504c83f..0cf3a6e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/GridComponent.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/GridComponent.java
@@ -64,7 +64,10 @@ public interface GridComponent {
QUERY_PROC,
/** Authentication processor. */
- AUTH_PROC
+ AUTH_PROC,
+
+ /** */
+ CACHE_CRD_PROC
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java
index 051978c..0690565 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java
@@ -32,6 +32,7 @@ import org.apache.ignite.internal.managers.eventstorage.GridEventStorageManager;
import org.apache.ignite.internal.managers.failover.GridFailoverManager;
import org.apache.ignite.internal.managers.indexing.GridIndexingManager;
import org.apache.ignite.internal.managers.loadbalancer.GridLoadBalancerManager;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccProcessor;
import org.apache.ignite.internal.worker.WorkersRegistry;
import org.apache.ignite.internal.processors.affinity.GridAffinityProcessor;
import org.apache.ignite.internal.processors.authentication.IgniteAuthenticationProcessor;
@@ -671,6 +672,11 @@ public interface GridKernalContext extends Iterable<GridComponent> {
public PlatformProcessor platform();
/**
+ * @return Cache mvcc coordinator processor.
+ */
+ public MvccProcessor coordinators();
+
+ /**
* @return PDS mode folder name resolver, also generates consistent ID in case new folder naming is used
*/
public PdsFoldersResolver pdsFolderResolver();
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
index 2be64e5..3b7b430 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
@@ -47,6 +47,7 @@ import org.apache.ignite.internal.managers.eventstorage.GridEventStorageManager;
import org.apache.ignite.internal.managers.failover.GridFailoverManager;
import org.apache.ignite.internal.managers.indexing.GridIndexingManager;
import org.apache.ignite.internal.managers.loadbalancer.GridLoadBalancerManager;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccProcessor;
import org.apache.ignite.internal.worker.WorkersRegistry;
import org.apache.ignite.internal.processors.affinity.GridAffinityProcessor;
import org.apache.ignite.internal.processors.authentication.IgniteAuthenticationProcessor;
@@ -290,6 +291,10 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable
@GridToStringExclude
private DataStructuresProcessor dataStructuresProc;
+ /** Cache mvcc coordinators. */
+ @GridToStringExclude
+ private MvccProcessor coordProc;
+
/** */
@GridToStringExclude
private IgniteAuthenticationProcessor authProc;
@@ -356,7 +361,7 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable
/** */
@GridToStringExclude
- Map<String, ? extends ExecutorService> customExecSvcs;
+ private Map<String, ? extends ExecutorService> customExecSvcs;
/** */
@GridToStringExclude
@@ -610,6 +615,8 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable
poolProc = (PoolProcessor)comp;
else if (comp instanceof GridMarshallerMappingProcessor)
mappingProc = (GridMarshallerMappingProcessor)comp;
+ else if (comp instanceof MvccProcessor)
+ coordProc = (MvccProcessor)comp;
else if (comp instanceof PdsFoldersResolver)
pdsFolderRslvr = (PdsFoldersResolver)comp;
else if (comp instanceof GridInternalSubscriptionProcessor)
@@ -880,6 +887,11 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable
}
/** {@inheritDoc} */
+ @Override public MvccProcessor coordinators() {
+ return coordProc;
+ }
+
+ /** {@inheritDoc} */
@Override public IgniteAuthenticationProcessor authentication() {
return authProc;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/GridTopic.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridTopic.java b/modules/core/src/main/java/org/apache/ignite/internal/GridTopic.java
index 0b2d41a..68cea4f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/GridTopic.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/GridTopic.java
@@ -127,7 +127,10 @@ public enum GridTopic {
TOPIC_AUTH,
/** */
- TOPIC_EXCHANGE;
+ TOPIC_EXCHANGE,
+
+ /** */
+ TOPIC_CACHE_COORDINATOR;
/** Enum values. */
private static final GridTopic[] VALS = values();
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
index eef2140..230c05c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
@@ -125,6 +125,7 @@ import org.apache.ignite.internal.processors.cache.GridCacheUtilityKey;
import org.apache.ignite.internal.processors.cache.IgniteCacheProxy;
import org.apache.ignite.internal.processors.cache.IgniteInternalCache;
import org.apache.ignite.internal.processors.cache.binary.CacheObjectBinaryProcessorImpl;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccUtils;
import org.apache.ignite.internal.processors.cache.persistence.DataRegion;
import org.apache.ignite.internal.processors.cache.persistence.filename.PdsConsistentIdProcessor;
import org.apache.ignite.internal.processors.cacheobject.IgniteCacheObjectProcessor;
@@ -993,6 +994,7 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable {
// be able to start receiving messages once discovery completes.
try {
startProcessor(new PdsConsistentIdProcessor(ctx));
+ startProcessor(MvccUtils.createProcessor(ctx));
startProcessor(createComponent(DiscoveryNodeValidationProcessor.class, ctx));
startProcessor(new GridAffinityProcessor(ctx));
startProcessor(createComponent(GridSegmentationProcessor.class, ctx));
@@ -2761,7 +2763,7 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable {
/** {@inheritDoc} */
@Override public <K, V> IgniteCache<K, V> createCache(CacheConfiguration<K, V> cacheCfg) {
A.notNull(cacheCfg, "cacheCfg");
- CU.validateCacheName(cacheCfg.getName());
+ CU.validateNewCacheName(cacheCfg.getName());
guard();
@@ -2817,7 +2819,7 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable {
/** {@inheritDoc} */
@Override public <K, V> IgniteCache<K, V> createCache(String cacheName) {
- CU.validateCacheName(cacheName);
+ CU.validateNewCacheName(cacheName);
guard();
@@ -2846,7 +2848,7 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable {
@Override public <K, V> IgniteBiTuple<IgniteCache<K, V>, Boolean> getOrCreateCache0(
CacheConfiguration<K, V> cacheCfg, boolean sql) {
A.notNull(cacheCfg, "cacheCfg");
- CU.validateCacheName(cacheCfg.getName());
+ CU.validateNewCacheName(cacheCfg.getName());
guard();
@@ -2912,7 +2914,7 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable {
NearCacheConfiguration<K, V> nearCfg
) {
A.notNull(cacheCfg, "cacheCfg");
- CU.validateCacheName(cacheCfg.getName());
+ CU.validateNewCacheName(cacheCfg.getName());
A.notNull(nearCfg, "nearCfg");
guard();
@@ -2941,7 +2943,7 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable {
@Override public <K, V> IgniteCache<K, V> getOrCreateCache(CacheConfiguration<K, V> cacheCfg,
NearCacheConfiguration<K, V> nearCfg) {
A.notNull(cacheCfg, "cacheCfg");
- CU.validateCacheName(cacheCfg.getName());
+ CU.validateNewCacheName(cacheCfg.getName());
A.notNull(nearCfg, "nearCfg");
guard();
@@ -2982,7 +2984,7 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable {
/** {@inheritDoc} */
@Override public <K, V> IgniteCache<K, V> createNearCache(String cacheName, NearCacheConfiguration<K, V> nearCfg) {
- CU.validateCacheName(cacheName);
+ CU.validateNewCacheName(cacheName);
A.notNull(nearCfg, "nearCfg");
guard();
@@ -3014,7 +3016,7 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable {
/** {@inheritDoc} */
@Override public <K, V> IgniteCache<K, V> getOrCreateNearCache(String cacheName,
NearCacheConfiguration<K, V> nearCfg) {
- CU.validateCacheName(cacheName);
+ CU.validateNewCacheName(cacheName);
A.notNull(nearCfg, "nearCfg");
guard();
@@ -3143,7 +3145,7 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable {
/** {@inheritDoc} */
@Override public <K, V> IgniteCache<K, V> getOrCreateCache(String cacheName) {
- CU.validateCacheName(cacheName);
+ CU.validateNewCacheName(cacheName);
guard();
@@ -3172,7 +3174,7 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable {
*/
public IgniteInternalFuture<?> getOrCreateCacheAsync(String cacheName, String templateName,
CacheConfigurationOverride cfgOverride, boolean checkThreadTx) {
- CU.validateCacheName(cacheName);
+ CU.validateNewCacheName(cacheName);
guard();
@@ -3192,7 +3194,7 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable {
/** {@inheritDoc} */
@Override public <K, V> void addCacheConfiguration(CacheConfiguration<K, V> cacheCfg) {
A.notNull(cacheCfg, "cacheCfg");
- CU.validateCacheName(cacheCfg.getName());
+ CU.validateNewCacheName(cacheCfg.getName());
guard();
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/IgniteNodeAttributes.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteNodeAttributes.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteNodeAttributes.java
index ed16a77..61a98a1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteNodeAttributes.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteNodeAttributes.java
@@ -205,6 +205,9 @@ public final class IgniteNodeAttributes {
/** Internal attribute name constant. */
public static final String ATTR_DYNAMIC_CACHE_START_ROLLBACK_SUPPORTED = ATTR_PREFIX + ".dynamic.cache.start.rollback.supported";
+ /** Mvcc enabled flag. */
+ public static final String ATTR_MVCC_ENABLED = ATTR_PREFIX + ".mvcc.enabled";
+
/**
* Enforces singleton.
*/
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java b/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java
index 2b408df..148bd21 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java
@@ -2349,12 +2349,8 @@ public class IgnitionEx {
"like TcpDiscoverySpi)");
for (CacheConfiguration ccfg : userCaches) {
- if (CU.isHadoopSystemCache(ccfg.getName()))
- throw new IgniteCheckedException("Cache name cannot be \"" + CU.SYS_CACHE_HADOOP_MR +
- "\" because it is reserved for internal purposes.");
-
- if (CU.isUtilityCache(ccfg.getName()))
- throw new IgniteCheckedException("Cache name cannot be \"" + CU.UTILITY_CACHE_NAME +
+ if (CU.isReservedCacheName(ccfg.getName()))
+ throw new IgniteCheckedException("Cache name cannot be \"" + ccfg.getName() +
"\" because it is reserved for internal purposes.");
if (IgfsUtils.matchIgfsCacheName(ccfg.getName()))
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/jdbc/thin/ConnectionProperties.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/jdbc/thin/ConnectionProperties.java b/modules/core/src/main/java/org/apache/ignite/internal/jdbc/thin/ConnectionProperties.java
index 92355a0..5e40bbf 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/jdbc/thin/ConnectionProperties.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/jdbc/thin/ConnectionProperties.java
@@ -164,6 +164,16 @@ public interface ConnectionProperties {
public void setSkipReducerOnUpdate(boolean skipReducerOnUpdate);
/**
+ * @return Nested transactions handling strategy.
+ */
+ public String nestedTxMode();
+
+ /**
+ * @param nestedTxMode Nested transactions handling strategy.
+ */
+ public void nestedTxMode(String nestedTxMode);
+
+ /**
* Gets SSL connection mode.
*
* @return Use SSL flag.
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/jdbc/thin/ConnectionPropertiesImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/jdbc/thin/ConnectionPropertiesImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/jdbc/thin/ConnectionPropertiesImpl.java
index 9371a07..51a3837 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/jdbc/thin/ConnectionPropertiesImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/jdbc/thin/ConnectionPropertiesImpl.java
@@ -27,6 +27,7 @@ import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.configuration.ClientConnectorConfiguration;
import org.apache.ignite.internal.processors.odbc.SqlStateCode;
import org.apache.ignite.internal.util.HostAndPortRange;
+import org.apache.ignite.internal.processors.query.NestedTxMode;
import org.apache.ignite.internal.util.typedef.F;
/**
@@ -99,6 +100,26 @@ public class ConnectionPropertiesImpl implements ConnectionProperties, Serializa
private BooleanProperty skipReducerOnUpdate = new BooleanProperty(
"skipReducerOnUpdate", "Enable execution update queries on ignite server nodes", false, false);
+ /** Nested transactions handling strategy. */
+ private StringProperty nestedTxMode = new StringProperty(
+ "nestedTransactionsMode", "Way to handle nested transactions", NestedTxMode.ERROR.name(),
+ new String[] { NestedTxMode.COMMIT.name(), NestedTxMode.ERROR.name(), NestedTxMode.IGNORE.name() },
+ false, new PropertyValidator() {
+ private static final long serialVersionUID = 0L;
+
+ @Override public void validate(String mode) throws SQLException {
+ if (!F.isEmpty(mode)) {
+ try {
+ NestedTxMode.valueOf(mode.toUpperCase());
+ }
+ catch (IllegalArgumentException e) {
+ throw new SQLException("Invalid nested transactions handling mode, allowed values: " +
+ Arrays.toString(nestedTxMode.choices), SqlStateCode.CLIENT_CONNECTION_FAILED);
+ }
+ }
+ }
+ });
+
/** SSL: Use SSL connection to Ignite node. */
private StringProperty sslMode = new StringProperty("sslMode",
"The SSL mode of the connection", SSL_MODE_DISABLE,
@@ -165,7 +186,7 @@ public class ConnectionPropertiesImpl implements ConnectionProperties, Serializa
/** Properties array. */
private final ConnectionProperty [] propsArray = {
distributedJoins, enforceJoinOrder, collocated, replicatedOnly, autoCloseServerCursor,
- tcpNoDelay, lazy, socketSendBuffer, socketReceiveBuffer, skipReducerOnUpdate,
+ tcpNoDelay, lazy, socketSendBuffer, socketReceiveBuffer, skipReducerOnUpdate, nestedTxMode,
sslMode, sslProtocol, sslKeyAlgorithm,
sslClientCertificateKeyStoreUrl, sslClientCertificateKeyStorePassword, sslClientCertificateKeyStoreType,
sslTrustCertificateKeyStoreUrl, sslTrustCertificateKeyStorePassword, sslTrustCertificateKeyStoreType,
@@ -437,6 +458,16 @@ public class ConnectionPropertiesImpl implements ConnectionProperties, Serializa
}
/** {@inheritDoc} */
+ @Override public String nestedTxMode() {
+ return nestedTxMode.value();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void nestedTxMode(String val) {
+ nestedTxMode.setValue(val);
+ }
+
+ /** {@inheritDoc} */
@Override public void setUsername(String name) {
user.setValue(name);
}
@@ -815,11 +846,11 @@ public class ConnectionPropertiesImpl implements ConnectionProperties, Serializa
SqlStateCode.CLIENT_CONNECTION_FAILED);
}
- checkChoices(strVal);
-
if (validator != null)
validator.validate(strVal);
+ checkChoices(strVal);
+
props.remove(name);
init(strVal);
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/jdbc/thin/JdbcThinConnection.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/jdbc/thin/JdbcThinConnection.java b/modules/core/src/main/java/org/apache/ignite/internal/jdbc/thin/JdbcThinConnection.java
index b36b319..547b1db 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/jdbc/thin/JdbcThinConnection.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/jdbc/thin/JdbcThinConnection.java
@@ -197,7 +197,7 @@ public class JdbcThinConnection implements Connection {
}
sendRequest(new JdbcQueryExecuteRequest(JdbcStatementType.ANY_STATEMENT_TYPE,
- schema, 1, 1, sql, null));
+ schema, 1, 1, autoCommit, sql, null));
streamState = new StreamState((SqlSetStreamingCommand)cmd);
}
@@ -328,19 +328,18 @@ public class JdbcThinConnection implements Connection {
@Override public void setAutoCommit(boolean autoCommit) throws SQLException {
ensureNotClosed();
- this.autoCommit = autoCommit;
+ // Do nothing if resulting value doesn't actually change.
+ if (autoCommit != this.autoCommit) {
+ doCommit();
- if (!autoCommit)
- LOG.warning("Transactions are not supported.");
+ this.autoCommit = autoCommit;
+ }
}
/** {@inheritDoc} */
@Override public boolean getAutoCommit() throws SQLException {
ensureNotClosed();
- if (!autoCommit)
- LOG.warning("Transactions are not supported.");
-
return autoCommit;
}
@@ -351,7 +350,7 @@ public class JdbcThinConnection implements Connection {
if (autoCommit)
throw new SQLException("Transaction cannot be committed explicitly in auto-commit mode.");
- LOG.warning("Transactions are not supported.");
+ doCommit();
}
/** {@inheritDoc} */
@@ -359,9 +358,21 @@ public class JdbcThinConnection implements Connection {
ensureNotClosed();
if (autoCommit)
- throw new SQLException("Transaction cannot rollback in auto-commit mode.");
+ throw new SQLException("Transaction cannot be rolled back explicitly in auto-commit mode.");
- LOG.warning("Transactions are not supported.");
+ try (Statement s = createStatement()) {
+ s.execute("ROLLBACK");
+ }
+ }
+
+ /**
+ * Send to the server {@code COMMIT} command.
+ * @throws SQLException if failed.
+ */
+ private void doCommit() throws SQLException {
+ try (Statement s = createStatement()) {
+ s.execute("COMMIT");
+ }
}
/** {@inheritDoc} */
@@ -927,7 +938,7 @@ public class JdbcThinConnection implements Connection {
respSem.acquire();
sendRequestNotWaitResponse(
- new JdbcOrderedBatchExecuteRequest(schema, streamBatch, lastBatch, order));
+ new JdbcOrderedBatchExecuteRequest(schema, streamBatch, autoCommit, lastBatch, order));
streamBatch = null;