zstan commented on code in PR #2846:
URL: https://github.com/apache/ignite-3/pull/2846#discussion_r1408797561
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/tx/ImplicitTransactionWrapper.java:
##########
@@ -24,38 +24,32 @@
/**
* Wrapper for the transaction that encapsulates the management of an implicit
transaction.
*/
-public class QueryTransactionWrapper {
+public class ImplicitTransactionWrapper implements QueryTransactionWrapper {
Review Comment:
naming supposed that this wrapper is for implicit tx, but semantically
(boolean implicit) it is not ) It confusing, change naming.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/BaseSqlMultiStatementTest.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.sql;
+
+import static
org.apache.ignite.internal.sql.engine.property.SqlPropertiesHelper.emptyProperties;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.await;
+import static
org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.empty;
+import static org.hamcrest.Matchers.equalTo;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import org.apache.ignite.internal.sql.engine.AsyncSqlCursor;
+import org.apache.ignite.internal.tx.InternalTransaction;
+import org.apache.ignite.internal.util.AsyncCursor.BatchedResult;
+import org.jetbrains.annotations.Nullable;
+import org.junit.jupiter.api.AfterEach;
+
+/**
+ * Base class for SQL multi-statement queries integration tests.
+ */
+public abstract class BaseSqlMultiStatementTest extends BaseSqlIntegrationTest
{
+ @AfterEach
+ protected void checkNoPendingTransactionsAndOpenedCursors() {
+ assertEquals(0, txManager().pending());
+
+ try {
+ boolean success = waitForCondition(() ->
queryProcessor().openedCursors() == 0, 5_000);
+
+ if (!success) {
+ assertEquals(0, queryProcessor().openedCursors());
Review Comment:
why do we need such additional check ?
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItSqlMultiStatementTxTest.java:
##########
@@ -0,0 +1,301 @@
+/*
+ * 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.sql.engine;
+
+import static
org.apache.ignite.internal.sql.engine.util.SqlTestUtils.assertThrowsSqlException;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.await;
+import static
org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.apache.ignite.lang.ErrorGroups.Sql.RUNTIME_ERR;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasSize;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrowsExactly;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+import java.util.List;
+import org.apache.ignite.internal.sql.BaseSqlMultiStatementTest;
+import org.apache.ignite.internal.sql.engine.util.Commons;
+import org.apache.ignite.internal.tx.InternalTransaction;
+import org.apache.ignite.internal.util.AsyncCursor.BatchedResult;
+import org.apache.ignite.sql.ExternalTransactionNotSupportedException;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests to verify the execution of queries with transaction control
statements.
+ *
+ * @see SqlQueryType#TX_CONTROL
+ */
+@SuppressWarnings("ThrowableNotThrown")
+public class ItSqlMultiStatementTxTest extends BaseSqlMultiStatementTest {
+ /** Default number of rows in the big table. */
+ private static final int BIG_TABLE_ROWS_COUNT = Commons.IN_BUFFER_SIZE * 6;
+
+ /** Number of completed transactions before the test started. */
+ private int numberOfFinishedTransactionsOnStart;
+
+ @BeforeAll
+ void createTable() {
+ sql("CREATE TABLE test (id INT PRIMARY KEY)");
+
+ createTable("big", 1, 1);
+ sql("INSERT INTO big (id) SELECT x FROM TABLE(SYSTEM_RANGE(1, " +
BIG_TABLE_ROWS_COUNT + "));");
+ }
+
+ @AfterEach
+ void clearTestTable() {
+ sql("DELETE FROM test");
+ }
+
+ @BeforeEach
+ void saveFinishedTxCount() {
+ numberOfFinishedTransactionsOnStart = txManager().finished();
+ }
+
+ @Test
+ void emptyTransactionControlStatement() {
+ List<AsyncSqlCursor<List<Object>>> cursors = fetchAllCursors(
+ runScript("START TRANSACTION; COMMIT"));
+
+ assertThat(cursors, hasSize(2));
+
+ validateSingleResult(cursors.get(0));
+ validateSingleResult(cursors.get(1));
+
+ verifyFinishedTxCount(1);
+ }
+
+ @Test
+ void basicTxStatements() {
+ List<AsyncSqlCursor<List<Object>>> cursors = fetchAllCursors(runScript(
+ "START TRANSACTION;"
+ + "INSERT INTO test VALUES(0);"
+ + "INSERT INTO test VALUES(1);"
+ + "COMMIT;"
+
+ + "START TRANSACTION;"
+ + "COMMIT;"
+
+ + "START TRANSACTION;"
+ + "INSERT INTO test VALUES(2);"
+ + "COMMIT;"
+ ));
+
+ verifyFinishedTxCount(3);
+
+ assertQuery("select count(id) from test")
+ .returns(3L).check();
+
+ cursors.forEach(AsyncSqlCursor::closeAsync);
+ }
+
+ @Test
+ void readMoreRowsThanCanBePrefetchedReadOnlyTx() {
+ String query = "START TRANSACTION READ ONLY;"
+ + "SELECT 1;"
+ + "SELECT id FROM big;"
+ + "COMMIT;";
+
+ List<AsyncSqlCursor<List<Object>>> cursors =
fetchAllCursors(runScript(query));
+ assertThat(cursors, hasSize(4));
+
+ BatchedResult<List<Object>> res =
await(cursors.get(2).requestNextAsync(BIG_TABLE_ROWS_COUNT * 2));
+ assertNotNull(res);
+ assertThat(res.items(), hasSize(BIG_TABLE_ROWS_COUNT));
+
+ verifyFinishedTxCount(1);
+
+ cursors.forEach(AsyncSqlCursor::closeAsync);
+ }
+
+ @Test
+ void readMoreRowsThanCanBePrefetchedReadWriteTx() {
Review Comment:
```suggestion
void readMoreRowsThanCanBePrefetchedReadWriteTx() {
String query = "START TRANSACTION;"
+ "UPDATE big SET salary=1 WHERE id=1;"
+ "SELECT id FROM big;"
+ "SELECT id FROM big;"
+ "COMMIT;";
AsyncSqlCursor<List<Object>> cursor = runScript(query);
assertTrue(cursor.hasNextResult());
AsyncSqlCursor<List<Object>> updateCursor =
await(cursor.nextResult());
assertNotNull(updateCursor);
validateSingleResult(updateCursor, 1L);
assertTrue(updateCursor.hasNextResult());
AsyncSqlCursor<List<Object>> selectCursor0 =
await(updateCursor.nextResult());
assertNotNull(selectCursor0);
BatchedResult<List<Object>> res = await(
selectCursor0.requestNextAsync(BIG_TABLE_ROWS_COUNT / 2));
assertNotNull(res);
assertThat(res.items(), hasSize(BIG_TABLE_ROWS_COUNT / 2));
assertEquals(1, txManager().pending(), "Transaction must not
finished until the cursor is closed.");
AsyncSqlCursor<List<Object>> selectCursor1 =
await(selectCursor0.nextResult());
res = await(
selectCursor1.requestNextAsync(BIG_TABLE_ROWS_COUNT * 2));
assertNotNull(res);
assertThat(res.items(), hasSize(BIG_TABLE_ROWS_COUNT));
assertEquals(1, txManager().pending(), "Transaction must not
finished until the cursor is closed.");
await(selectCursor0.requestNextAsync(BIG_TABLE_ROWS_COUNT)); //
Cursor must close implicitly.
assertFalse(res.hasMore());
verifyFinishedTxCount(1);
}
```
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItSqlMultiStatementTxTest.java:
##########
@@ -0,0 +1,301 @@
+/*
+ * 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.sql.engine;
+
+import static
org.apache.ignite.internal.sql.engine.util.SqlTestUtils.assertThrowsSqlException;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.await;
+import static
org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.apache.ignite.lang.ErrorGroups.Sql.RUNTIME_ERR;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasSize;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrowsExactly;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+import java.util.List;
+import org.apache.ignite.internal.sql.BaseSqlMultiStatementTest;
+import org.apache.ignite.internal.sql.engine.util.Commons;
+import org.apache.ignite.internal.tx.InternalTransaction;
+import org.apache.ignite.internal.util.AsyncCursor.BatchedResult;
+import org.apache.ignite.sql.ExternalTransactionNotSupportedException;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests to verify the execution of queries with transaction control
statements.
+ *
+ * @see SqlQueryType#TX_CONTROL
+ */
+@SuppressWarnings("ThrowableNotThrown")
+public class ItSqlMultiStatementTxTest extends BaseSqlMultiStatementTest {
+ /** Default number of rows in the big table. */
+ private static final int BIG_TABLE_ROWS_COUNT = Commons.IN_BUFFER_SIZE * 6;
+
+ /** Number of completed transactions before the test started. */
+ private int numberOfFinishedTransactionsOnStart;
+
+ @BeforeAll
+ void createTable() {
+ sql("CREATE TABLE test (id INT PRIMARY KEY)");
+
+ createTable("big", 1, 1);
+ sql("INSERT INTO big (id) SELECT x FROM TABLE(SYSTEM_RANGE(1, " +
BIG_TABLE_ROWS_COUNT + "));");
+ }
+
+ @AfterEach
+ void clearTestTable() {
+ sql("DELETE FROM test");
+ }
+
+ @BeforeEach
+ void saveFinishedTxCount() {
+ numberOfFinishedTransactionsOnStart = txManager().finished();
+ }
+
+ @Test
+ void emptyTransactionControlStatement() {
+ List<AsyncSqlCursor<List<Object>>> cursors = fetchAllCursors(
+ runScript("START TRANSACTION; COMMIT"));
+
+ assertThat(cursors, hasSize(2));
+
+ validateSingleResult(cursors.get(0));
+ validateSingleResult(cursors.get(1));
+
+ verifyFinishedTxCount(1);
+ }
+
+ @Test
+ void basicTxStatements() {
+ List<AsyncSqlCursor<List<Object>>> cursors = fetchAllCursors(runScript(
+ "START TRANSACTION;"
+ + "INSERT INTO test VALUES(0);"
+ + "INSERT INTO test VALUES(1);"
+ + "COMMIT;"
+
+ + "START TRANSACTION;"
+ + "COMMIT;"
+
+ + "START TRANSACTION;"
+ + "INSERT INTO test VALUES(2);"
+ + "COMMIT;"
+ ));
+
+ verifyFinishedTxCount(3);
+
+ assertQuery("select count(id) from test")
+ .returns(3L).check();
+
+ cursors.forEach(AsyncSqlCursor::closeAsync);
+ }
+
+ @Test
+ void readMoreRowsThanCanBePrefetchedReadOnlyTx() {
+ String query = "START TRANSACTION READ ONLY;"
+ + "SELECT 1;"
+ + "SELECT id FROM big;"
+ + "COMMIT;";
+
+ List<AsyncSqlCursor<List<Object>>> cursors =
fetchAllCursors(runScript(query));
+ assertThat(cursors, hasSize(4));
+
+ BatchedResult<List<Object>> res =
await(cursors.get(2).requestNextAsync(BIG_TABLE_ROWS_COUNT * 2));
+ assertNotNull(res);
+ assertThat(res.items(), hasSize(BIG_TABLE_ROWS_COUNT));
+
+ verifyFinishedTxCount(1);
+
+ cursors.forEach(AsyncSqlCursor::closeAsync);
+ }
+
+ @Test
+ void readMoreRowsThanCanBePrefetchedReadWriteTx() {
+ String query = "START TRANSACTION;"
+ + "UPDATE big SET salary=1 WHERE id=1;"
+ + "SELECT id FROM big;"
+ + "COMMIT;";
+
+ AsyncSqlCursor<List<Object>> cursor = runScript(query);
+
+ assertTrue(cursor.hasNextResult());
+
+ AsyncSqlCursor<List<Object>> updateCursor = await(cursor.nextResult());
+ assertNotNull(updateCursor);
+ validateSingleResult(updateCursor, 1L);
+ assertTrue(updateCursor.hasNextResult());
+
+ AsyncSqlCursor<List<Object>> selectCursor =
await(updateCursor.nextResult());
+ assertNotNull(selectCursor);
+
+ BatchedResult<List<Object>> res = await(
+ selectCursor.requestNextAsync(BIG_TABLE_ROWS_COUNT / 2));
+
+ assertNotNull(res);
+ assertThat(res.items(), hasSize(BIG_TABLE_ROWS_COUNT / 2));
+ assertEquals(1, txManager().pending(), "Transaction must not finished
until the cursor is closed.");
+ assertFalse(selectCursor.nextResult().isDone());
+
+ await(selectCursor.requestNextAsync(BIG_TABLE_ROWS_COUNT)); // Cursor
must close implicitly.
+ verifyFinishedTxCount(1);
+ }
+
+ @Test
+ void openedScriptTransactionRollsBackImplicitly() {
+ {
+ fetchAllCursors(runScript("START TRANSACTION;"));
+ verifyFinishedTxCount(1);
+ }
+
+ {
+ List<AsyncSqlCursor<List<Object>>> cursors = fetchAllCursors(
+ runScript("START TRANSACTION;"
+ + "INSERT INTO test VALUES(0);"
+ + "INSERT INTO test VALUES(1);"
+ + "COMMIT;"
+
+ + "START TRANSACTION;"
+ + "INSERT INTO test VALUES(2);"
+ + "SELECT * FROM test;"
+ )
+ );
+
+ verifyFinishedTxCount(3);
Review Comment:
we need to document it, and check that such decision is responds to referral
data bases
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/tx/ScriptTransactionHandler.java:
##########
@@ -0,0 +1,319 @@
+/*
+ * 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.sql.engine.tx;
+
+import static org.apache.ignite.lang.ErrorGroups.Sql.RUNTIME_ERR;
+
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.ignite.internal.sql.engine.SqlQueryType;
+import org.apache.ignite.internal.sql.engine.sql.IgniteSqlCommitTransaction;
+import org.apache.ignite.internal.sql.engine.sql.IgniteSqlStartTransaction;
+import org.apache.ignite.internal.sql.engine.sql.IgniteSqlStartTransactionMode;
+import org.apache.ignite.internal.sql.engine.sql.ParsedResult;
+import org.apache.ignite.internal.sql.engine.util.Commons;
+import org.apache.ignite.internal.tx.InternalTransaction;
+import org.apache.ignite.internal.util.AsyncCursor;
+import org.apache.ignite.sql.ExternalTransactionNotSupportedException;
+import org.apache.ignite.sql.SqlException;
+import org.apache.ignite.tx.IgniteTransactions;
+import org.apache.ignite.tx.TransactionOptions;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Starts an implicit transaction if there is no external transaction.
Supports script transaction management using
+ * {@link SqlQueryType#TX_CONTROL} statements.
+ */
+public class ScriptTransactionHandler extends QueryTransactionHandler {
+ /** No-op transaction wrapper. */
+ private static final QueryTransactionWrapper NOOP_TX_WRAPPER = new
NoopTransactionWrapper();
+
+ /** Wraps a transaction, which is managed by SQL engine via {@link
SqlQueryType#TX_CONTROL} statements. */
+ private volatile @Nullable ManagedTransactionWrapper wrapper;
+
+ public ScriptTransactionHandler(IgniteTransactions transactions, @Nullable
InternalTransaction externalTransaction) {
Review Comment:
InternalTransaction externalTransaction - sounds weird )
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItSqlMultiStatementTxTest.java:
##########
@@ -0,0 +1,301 @@
+/*
+ * 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.sql.engine;
+
+import static
org.apache.ignite.internal.sql.engine.util.SqlTestUtils.assertThrowsSqlException;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.await;
+import static
org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.apache.ignite.lang.ErrorGroups.Sql.RUNTIME_ERR;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasSize;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrowsExactly;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+import java.util.List;
+import org.apache.ignite.internal.sql.BaseSqlMultiStatementTest;
+import org.apache.ignite.internal.sql.engine.util.Commons;
+import org.apache.ignite.internal.tx.InternalTransaction;
+import org.apache.ignite.internal.util.AsyncCursor.BatchedResult;
+import org.apache.ignite.sql.ExternalTransactionNotSupportedException;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests to verify the execution of queries with transaction control
statements.
+ *
+ * @see SqlQueryType#TX_CONTROL
+ */
+@SuppressWarnings("ThrowableNotThrown")
+public class ItSqlMultiStatementTxTest extends BaseSqlMultiStatementTest {
+ /** Default number of rows in the big table. */
+ private static final int BIG_TABLE_ROWS_COUNT = Commons.IN_BUFFER_SIZE * 6;
+
+ /** Number of completed transactions before the test started. */
+ private int numberOfFinishedTransactionsOnStart;
+
+ @BeforeAll
+ void createTable() {
+ sql("CREATE TABLE test (id INT PRIMARY KEY)");
+
+ createTable("big", 1, 1);
+ sql("INSERT INTO big (id) SELECT x FROM TABLE(SYSTEM_RANGE(1, " +
BIG_TABLE_ROWS_COUNT + "));");
+ }
+
+ @AfterEach
+ void clearTestTable() {
+ sql("DELETE FROM test");
+ }
+
+ @BeforeEach
+ void saveFinishedTxCount() {
+ numberOfFinishedTransactionsOnStart = txManager().finished();
+ }
+
+ @Test
+ void emptyTransactionControlStatement() {
+ List<AsyncSqlCursor<List<Object>>> cursors = fetchAllCursors(
+ runScript("START TRANSACTION; COMMIT"));
+
+ assertThat(cursors, hasSize(2));
+
+ validateSingleResult(cursors.get(0));
+ validateSingleResult(cursors.get(1));
+
+ verifyFinishedTxCount(1);
+ }
+
+ @Test
+ void basicTxStatements() {
+ List<AsyncSqlCursor<List<Object>>> cursors = fetchAllCursors(runScript(
+ "START TRANSACTION;"
+ + "INSERT INTO test VALUES(0);"
+ + "INSERT INTO test VALUES(1);"
+ + "COMMIT;"
+
+ + "START TRANSACTION;"
+ + "COMMIT;"
+
+ + "START TRANSACTION;"
+ + "INSERT INTO test VALUES(2);"
+ + "COMMIT;"
+ ));
+
+ verifyFinishedTxCount(3);
+
+ assertQuery("select count(id) from test")
+ .returns(3L).check();
+
+ cursors.forEach(AsyncSqlCursor::closeAsync);
+ }
+
+ @Test
+ void readMoreRowsThanCanBePrefetchedReadOnlyTx() {
+ String query = "START TRANSACTION READ ONLY;"
+ + "SELECT 1;"
+ + "SELECT id FROM big;"
+ + "COMMIT;";
+
+ List<AsyncSqlCursor<List<Object>>> cursors =
fetchAllCursors(runScript(query));
+ assertThat(cursors, hasSize(4));
+
+ BatchedResult<List<Object>> res =
await(cursors.get(2).requestNextAsync(BIG_TABLE_ROWS_COUNT * 2));
+ assertNotNull(res);
+ assertThat(res.items(), hasSize(BIG_TABLE_ROWS_COUNT));
+
+ verifyFinishedTxCount(1);
+
+ cursors.forEach(AsyncSqlCursor::closeAsync);
+ }
+
+ @Test
+ void readMoreRowsThanCanBePrefetchedReadWriteTx() {
+ String query = "START TRANSACTION;"
+ + "UPDATE big SET salary=1 WHERE id=1;"
+ + "SELECT id FROM big;"
+ + "COMMIT;";
+
+ AsyncSqlCursor<List<Object>> cursor = runScript(query);
+
+ assertTrue(cursor.hasNextResult());
+
+ AsyncSqlCursor<List<Object>> updateCursor = await(cursor.nextResult());
+ assertNotNull(updateCursor);
+ validateSingleResult(updateCursor, 1L);
+ assertTrue(updateCursor.hasNextResult());
+
+ AsyncSqlCursor<List<Object>> selectCursor =
await(updateCursor.nextResult());
+ assertNotNull(selectCursor);
+
+ BatchedResult<List<Object>> res = await(
+ selectCursor.requestNextAsync(BIG_TABLE_ROWS_COUNT / 2));
+
+ assertNotNull(res);
+ assertThat(res.items(), hasSize(BIG_TABLE_ROWS_COUNT / 2));
+ assertEquals(1, txManager().pending(), "Transaction must not finished
until the cursor is closed.");
+ assertFalse(selectCursor.nextResult().isDone());
+
+ await(selectCursor.requestNextAsync(BIG_TABLE_ROWS_COUNT)); // Cursor
must close implicitly.
+ verifyFinishedTxCount(1);
+ }
+
+ @Test
+ void openedScriptTransactionRollsBackImplicitly() {
+ {
+ fetchAllCursors(runScript("START TRANSACTION;"));
+ verifyFinishedTxCount(1);
+ }
+
+ {
+ List<AsyncSqlCursor<List<Object>>> cursors = fetchAllCursors(
+ runScript("START TRANSACTION;"
+ + "INSERT INTO test VALUES(0);"
+ + "INSERT INTO test VALUES(1);"
+ + "COMMIT;"
+
+ + "START TRANSACTION;"
+ + "INSERT INTO test VALUES(2);"
+ + "SELECT * FROM test;"
+ )
+ );
+
+ verifyFinishedTxCount(3);
+
+ assertThat(cursors, hasSize(7));
+ cursors.subList(0, 4).forEach(AsyncSqlCursor::closeAsync);
+
+ // Cursors associated with a transaction must be closed when the
transaction is rolled back.
+ checkNoPendingTransactionsAndOpenedCursors();
+
+ assertQuery("select count(id) from test")
+ .returns(2L).check();
+ }
+ }
+
+ @Test
+ void openedScriptTransactionRollsBackOnError() {
+ {
+ AsyncSqlCursor<List<Object>> cursor = runScript(
+ "START TRANSACTION READ WRITE;"
+ + "INSERT INTO test VALUES(2);"
+ + "INSERT INTO test VALUES(2/0);"
+ + "SELECT 1;"
+ + "COMMIT;"
+ );
+
+ assertThrowsSqlException(RUNTIME_ERR, "/ by zero", () ->
fetchAllCursors(cursor));
Review Comment:
plz fill the issue for align error message with referral data bases ?
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItSqlMultiStatementTxTest.java:
##########
@@ -0,0 +1,301 @@
+/*
+ * 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.sql.engine;
+
+import static
org.apache.ignite.internal.sql.engine.util.SqlTestUtils.assertThrowsSqlException;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.await;
+import static
org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.apache.ignite.lang.ErrorGroups.Sql.RUNTIME_ERR;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasSize;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrowsExactly;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+import java.util.List;
+import org.apache.ignite.internal.sql.BaseSqlMultiStatementTest;
+import org.apache.ignite.internal.sql.engine.util.Commons;
+import org.apache.ignite.internal.tx.InternalTransaction;
+import org.apache.ignite.internal.util.AsyncCursor.BatchedResult;
+import org.apache.ignite.sql.ExternalTransactionNotSupportedException;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests to verify the execution of queries with transaction control
statements.
+ *
+ * @see SqlQueryType#TX_CONTROL
+ */
+@SuppressWarnings("ThrowableNotThrown")
+public class ItSqlMultiStatementTxTest extends BaseSqlMultiStatementTest {
+ /** Default number of rows in the big table. */
+ private static final int BIG_TABLE_ROWS_COUNT = Commons.IN_BUFFER_SIZE * 6;
+
+ /** Number of completed transactions before the test started. */
+ private int numberOfFinishedTransactionsOnStart;
+
+ @BeforeAll
+ void createTable() {
+ sql("CREATE TABLE test (id INT PRIMARY KEY)");
+
+ createTable("big", 1, 1);
+ sql("INSERT INTO big (id) SELECT x FROM TABLE(SYSTEM_RANGE(1, " +
BIG_TABLE_ROWS_COUNT + "));");
+ }
+
+ @AfterEach
+ void clearTestTable() {
+ sql("DELETE FROM test");
+ }
+
+ @BeforeEach
+ void saveFinishedTxCount() {
+ numberOfFinishedTransactionsOnStart = txManager().finished();
+ }
+
+ @Test
+ void emptyTransactionControlStatement() {
+ List<AsyncSqlCursor<List<Object>>> cursors = fetchAllCursors(
+ runScript("START TRANSACTION; COMMIT"));
+
+ assertThat(cursors, hasSize(2));
+
+ validateSingleResult(cursors.get(0));
+ validateSingleResult(cursors.get(1));
+
+ verifyFinishedTxCount(1);
+ }
+
+ @Test
+ void basicTxStatements() {
+ List<AsyncSqlCursor<List<Object>>> cursors = fetchAllCursors(runScript(
+ "START TRANSACTION;"
+ + "INSERT INTO test VALUES(0);"
+ + "INSERT INTO test VALUES(1);"
+ + "COMMIT;"
+
+ + "START TRANSACTION;"
+ + "COMMIT;"
+
+ + "START TRANSACTION;"
+ + "INSERT INTO test VALUES(2);"
+ + "COMMIT;"
+ ));
+
+ verifyFinishedTxCount(3);
+
+ assertQuery("select count(id) from test")
+ .returns(3L).check();
+
+ cursors.forEach(AsyncSqlCursor::closeAsync);
+ }
+
+ @Test
+ void readMoreRowsThanCanBePrefetchedReadOnlyTx() {
+ String query = "START TRANSACTION READ ONLY;"
+ + "SELECT 1;"
+ + "SELECT id FROM big;"
+ + "COMMIT;";
+
+ List<AsyncSqlCursor<List<Object>>> cursors =
fetchAllCursors(runScript(query));
+ assertThat(cursors, hasSize(4));
+
+ BatchedResult<List<Object>> res =
await(cursors.get(2).requestNextAsync(BIG_TABLE_ROWS_COUNT * 2));
+ assertNotNull(res);
+ assertThat(res.items(), hasSize(BIG_TABLE_ROWS_COUNT));
+
+ verifyFinishedTxCount(1);
+
+ cursors.forEach(AsyncSqlCursor::closeAsync);
+ }
+
+ @Test
+ void readMoreRowsThanCanBePrefetchedReadWriteTx() {
+ String query = "START TRANSACTION;"
+ + "UPDATE big SET salary=1 WHERE id=1;"
+ + "SELECT id FROM big;"
+ + "COMMIT;";
+
+ AsyncSqlCursor<List<Object>> cursor = runScript(query);
+
+ assertTrue(cursor.hasNextResult());
+
+ AsyncSqlCursor<List<Object>> updateCursor = await(cursor.nextResult());
+ assertNotNull(updateCursor);
+ validateSingleResult(updateCursor, 1L);
+ assertTrue(updateCursor.hasNextResult());
+
+ AsyncSqlCursor<List<Object>> selectCursor =
await(updateCursor.nextResult());
+ assertNotNull(selectCursor);
+
+ BatchedResult<List<Object>> res = await(
+ selectCursor.requestNextAsync(BIG_TABLE_ROWS_COUNT / 2));
+
+ assertNotNull(res);
+ assertThat(res.items(), hasSize(BIG_TABLE_ROWS_COUNT / 2));
+ assertEquals(1, txManager().pending(), "Transaction must not finished
until the cursor is closed.");
+ assertFalse(selectCursor.nextResult().isDone());
+
+ await(selectCursor.requestNextAsync(BIG_TABLE_ROWS_COUNT)); // Cursor
must close implicitly.
+ verifyFinishedTxCount(1);
+ }
+
+ @Test
+ void openedScriptTransactionRollsBackImplicitly() {
+ {
+ fetchAllCursors(runScript("START TRANSACTION;"));
+ verifyFinishedTxCount(1);
+ }
+
+ {
+ List<AsyncSqlCursor<List<Object>>> cursors = fetchAllCursors(
+ runScript("START TRANSACTION;"
+ + "INSERT INTO test VALUES(0);"
+ + "INSERT INTO test VALUES(1);"
+ + "COMMIT;"
+
+ + "START TRANSACTION;"
+ + "INSERT INTO test VALUES(2);"
+ + "SELECT * FROM test;"
+ )
+ );
+
+ verifyFinishedTxCount(3);
+
+ assertThat(cursors, hasSize(7));
+ cursors.subList(0, 4).forEach(AsyncSqlCursor::closeAsync);
+
+ // Cursors associated with a transaction must be closed when the
transaction is rolled back.
+ checkNoPendingTransactionsAndOpenedCursors();
+
+ assertQuery("select count(id) from test")
+ .returns(2L).check();
+ }
+ }
+
+ @Test
+ void openedScriptTransactionRollsBackOnError() {
+ {
+ AsyncSqlCursor<List<Object>> cursor = runScript(
+ "START TRANSACTION READ WRITE;"
+ + "INSERT INTO test VALUES(2);"
+ + "INSERT INTO test VALUES(2/0);"
+ + "SELECT 1;"
+ + "COMMIT;"
+ );
+
+ assertThrowsSqlException(RUNTIME_ERR, "/ by zero", () ->
fetchAllCursors(cursor));
+
+ verifyFinishedTxCount(1);
+
+ assertQuery("select count(id) from test")
+ .returns(0L).check();
+ }
+ }
+
+ @Test
+ void commitWithoutOpenTransactionDoesNothing() {
+ List<AsyncSqlCursor<List<Object>>> cursors = fetchAllCursors(
+ runScript("COMMIT; COMMIT; COMMIT;"));
+
+ assertThat(cursors, hasSize(3));
+
+ verifyFinishedTxCount(0);
+ }
+
+ @Test
+ void ddlInsideExplicitTransactionFails() {
+ String ddlStatement = "CREATE TABLE foo (id INT PRIMARY KEY)";
+
+ {
+ InternalTransaction tx = (InternalTransaction) igniteTx().begin();
+
+ assertThrowsSqlException(RUNTIME_ERR, "DDL doesn't support
transactions.",
+ () -> runScript(ddlStatement, tx));
+
+ assertEquals(1, txManager().pending());
+ tx.rollback();
+
+ verifyFinishedTxCount(1);
+ }
+
+ {
+ assertThrowsSqlException(RUNTIME_ERR, "DDL doesn't support
transactions.",
+ () -> fetchAllCursors(runScript("START TRANSACTION;" +
ddlStatement)));
+
+ verifyFinishedTxCount(2);
+ }
+ }
+
+ @Test
+ void nestedTransactionStartFails() {
+ AsyncSqlCursor<List<Object>> cursor = runScript("START TRANSACTION;
START TRANSACTION;");
+ assertThrowsSqlException(RUNTIME_ERR, "Nested transactions are not
supported.", () -> await(cursor.nextResult()));
+
+ verifyFinishedTxCount(1);
+ }
+
+ @Test
+ void dmlFailsOnReadOnlyTransaction() {
+ AsyncSqlCursor<List<Object>> cursor = runScript("START TRANSACTION
READ ONLY;"
+ + "INSERT INTO test VALUES(0);"
+ + "COMMIT;");
+
+ assertThrowsSqlException(RUNTIME_ERR, "DML query cannot be started by
using read only transactions.",
+ () -> await(cursor.nextResult()));
+
+ verifyFinishedTxCount(1);
+ }
+
+ @Test
+ void transactionControlStatementFailsWithExternalTransaction() {
+ InternalTransaction tx1 = (InternalTransaction) igniteTx().begin();
+ assertThrowsExactly(ExternalTransactionNotSupportedException.class, ()
-> runScript("COMMIT", tx1));
+ assertEquals(1, txManager().pending());
+ tx1.rollback();
+
+ InternalTransaction tx2 = (InternalTransaction) igniteTx().begin();
+ assertThrowsExactly(ExternalTransactionNotSupportedException.class, ()
-> runScript("START TRANSACTION", tx2));
+ assertEquals(1, txManager().pending());
+ tx2.rollback();
+
+ verifyFinishedTxCount(2);
+ }
+
+ private void verifyFinishedTxCount(int expected) {
+ int expectedTotal = numberOfFinishedTransactionsOnStart + expected;
+
+ boolean success;
+
+ try {
+ success = waitForCondition(() -> expectedTotal ==
txManager().finished(), 2_000);
Review Comment:
```suggestion
assertTrue(waitForCondition(() -> expectedTotal ==
txManager().finished(), 2_000));
```
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItSqlMultiStatementTxTest.java:
##########
@@ -0,0 +1,301 @@
+/*
+ * 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.sql.engine;
+
+import static
org.apache.ignite.internal.sql.engine.util.SqlTestUtils.assertThrowsSqlException;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.await;
+import static
org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.apache.ignite.lang.ErrorGroups.Sql.RUNTIME_ERR;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasSize;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrowsExactly;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+import java.util.List;
+import org.apache.ignite.internal.sql.BaseSqlMultiStatementTest;
+import org.apache.ignite.internal.sql.engine.util.Commons;
+import org.apache.ignite.internal.tx.InternalTransaction;
+import org.apache.ignite.internal.util.AsyncCursor.BatchedResult;
+import org.apache.ignite.sql.ExternalTransactionNotSupportedException;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests to verify the execution of queries with transaction control
statements.
+ *
+ * @see SqlQueryType#TX_CONTROL
+ */
+@SuppressWarnings("ThrowableNotThrown")
+public class ItSqlMultiStatementTxTest extends BaseSqlMultiStatementTest {
+ /** Default number of rows in the big table. */
+ private static final int BIG_TABLE_ROWS_COUNT = Commons.IN_BUFFER_SIZE * 6;
+
+ /** Number of completed transactions before the test started. */
+ private int numberOfFinishedTransactionsOnStart;
+
+ @BeforeAll
+ void createTable() {
+ sql("CREATE TABLE test (id INT PRIMARY KEY)");
+
+ createTable("big", 1, 1);
+ sql("INSERT INTO big (id) SELECT x FROM TABLE(SYSTEM_RANGE(1, " +
BIG_TABLE_ROWS_COUNT + "));");
+ }
+
+ @AfterEach
+ void clearTestTable() {
+ sql("DELETE FROM test");
+ }
+
+ @BeforeEach
+ void saveFinishedTxCount() {
+ numberOfFinishedTransactionsOnStart = txManager().finished();
+ }
+
+ @Test
+ void emptyTransactionControlStatement() {
+ List<AsyncSqlCursor<List<Object>>> cursors = fetchAllCursors(
+ runScript("START TRANSACTION; COMMIT"));
+
+ assertThat(cursors, hasSize(2));
+
+ validateSingleResult(cursors.get(0));
+ validateSingleResult(cursors.get(1));
+
+ verifyFinishedTxCount(1);
+ }
+
+ @Test
+ void basicTxStatements() {
+ List<AsyncSqlCursor<List<Object>>> cursors = fetchAllCursors(runScript(
+ "START TRANSACTION;"
+ + "INSERT INTO test VALUES(0);"
+ + "INSERT INTO test VALUES(1);"
+ + "COMMIT;"
+
+ + "START TRANSACTION;"
+ + "COMMIT;"
+
+ + "START TRANSACTION;"
+ + "INSERT INTO test VALUES(2);"
+ + "COMMIT;"
+ ));
+
+ verifyFinishedTxCount(3);
+
+ assertQuery("select count(id) from test")
+ .returns(3L).check();
+
+ cursors.forEach(AsyncSqlCursor::closeAsync);
+ }
+
+ @Test
+ void readMoreRowsThanCanBePrefetchedReadOnlyTx() {
+ String query = "START TRANSACTION READ ONLY;"
+ + "SELECT 1;"
+ + "SELECT id FROM big;"
+ + "COMMIT;";
+
+ List<AsyncSqlCursor<List<Object>>> cursors =
fetchAllCursors(runScript(query));
+ assertThat(cursors, hasSize(4));
+
+ BatchedResult<List<Object>> res =
await(cursors.get(2).requestNextAsync(BIG_TABLE_ROWS_COUNT * 2));
+ assertNotNull(res);
+ assertThat(res.items(), hasSize(BIG_TABLE_ROWS_COUNT));
+
+ verifyFinishedTxCount(1);
+
+ cursors.forEach(AsyncSqlCursor::closeAsync);
+ }
+
+ @Test
+ void readMoreRowsThanCanBePrefetchedReadWriteTx() {
+ String query = "START TRANSACTION;"
+ + "UPDATE big SET salary=1 WHERE id=1;"
+ + "SELECT id FROM big;"
+ + "COMMIT;";
+
+ AsyncSqlCursor<List<Object>> cursor = runScript(query);
+
+ assertTrue(cursor.hasNextResult());
+
+ AsyncSqlCursor<List<Object>> updateCursor = await(cursor.nextResult());
+ assertNotNull(updateCursor);
+ validateSingleResult(updateCursor, 1L);
+ assertTrue(updateCursor.hasNextResult());
+
+ AsyncSqlCursor<List<Object>> selectCursor =
await(updateCursor.nextResult());
+ assertNotNull(selectCursor);
+
+ BatchedResult<List<Object>> res = await(
+ selectCursor.requestNextAsync(BIG_TABLE_ROWS_COUNT / 2));
+
+ assertNotNull(res);
+ assertThat(res.items(), hasSize(BIG_TABLE_ROWS_COUNT / 2));
+ assertEquals(1, txManager().pending(), "Transaction must not finished
until the cursor is closed.");
+ assertFalse(selectCursor.nextResult().isDone());
+
+ await(selectCursor.requestNextAsync(BIG_TABLE_ROWS_COUNT)); // Cursor
must close implicitly.
+ verifyFinishedTxCount(1);
+ }
+
+ @Test
+ void openedScriptTransactionRollsBackImplicitly() {
+ {
+ fetchAllCursors(runScript("START TRANSACTION;"));
+ verifyFinishedTxCount(1);
+ }
+
+ {
+ List<AsyncSqlCursor<List<Object>>> cursors = fetchAllCursors(
+ runScript("START TRANSACTION;"
+ + "INSERT INTO test VALUES(0);"
+ + "INSERT INTO test VALUES(1);"
+ + "COMMIT;"
+
Review Comment:
```suggestion
+ ";;"
```
lets additionally check unnecessary dividers ?
##########
modules/api/src/main/java/org/apache/ignite/sql/ExternalTransactionNotSupportedException.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * 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.sql;
+
+import java.util.UUID;
+import org.apache.ignite.lang.ErrorGroups.Sql;
+
+/**
+ * The exception is thrown when an attempt is made to execute a transaction
control statement
+ * ({@code START TRANSACTION} or {@code COMMIT}) while an external transaction
is already running.
+ */
+public class ExternalTransactionNotSupportedException extends SqlException {
+ /**
+ * Default constructor.
+ */
+ public ExternalTransactionNotSupportedException() {
+ super(Sql.TX_CONTROL_INSIDE_EXTERNAL_TX_ERR, "Transaction control
statement cannot be executed within an external transaction.");
+ }
+
+ /**
+ * Constructor is required to copy this exception.
+ */
+ @Deprecated
Review Comment:
1. seems not used
2. not "born" but already @Deprecated )?
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItSqlMultiStatementTest.java:
##########
@@ -138,131 +132,73 @@ void
queryWithIncorrectNumberOfDynamicParametersFailsWithValidationError() {
assertThrowsSqlException(STMT_VALIDATION_ERR, expectedMessage, () ->
runScript(sql, null, 0, 1, 2, 3, 4, 5));
}
- @Test
- void transactionControlStatementDoesNotCreateCursor() {
- assertThat(runScript("START TRANSACTION; COMMIT"), nullValue());
-
- AsyncSqlCursor<List<Object>> cursor = runScript(
- "START TRANSACTION;"
- + "SELECT 1;"
- + "COMMIT"
- );
-
- assertNotNull(cursor);
- validateSingleResult(cursor, 1);
-
- assertFalse(cursor.hasNextResult());
- }
-
@Test
void scriptStopsExecutionOnError() {
// Runtime error.
- AsyncSqlCursor<List<Object>> cursor = runScript(
- "CREATE TABLE test (id INT PRIMARY KEY);"
- + "SELECT 2/0;" // Runtime error.
- + "INSERT INTO test VALUES (0)"
- );
- assertNotNull(cursor);
- assertTrue(cursor.hasNextResult());
-
- CompletableFuture<AsyncSqlCursor<List<Object>>> curFut0 =
cursor.nextResult();
- assertThrowsSqlException(RUNTIME_ERR, "/ by zero", () ->
await(curFut0));
+ {
+ AsyncSqlCursor<List<Object>> cursor = runScript(
+ "CREATE TABLE test (id INT PRIMARY KEY);"
+ + "SELECT 2/0;" // Runtime error.
+ + "INSERT INTO test VALUES (0)"
+ );
+ assertTrue(cursor.hasNextResult());
Review Comment:
can you explain why :
```
List<AsyncSqlCursor<List<Object>>> cursors =
fetchAllCursors(runScript(
"CREATE TABLE test (id INT PRIMARY KEY);"
));
Iterator<AsyncSqlCursor<List<Object>>> curItr =
cursors.iterator();
validateSingleResult(curItr.next(), true);
```
works fine, but :
```
List<AsyncSqlCursor<List<Object>>> cursors =
fetchAllCursors(runScript(
"CREATE TABLE test (id INT PRIMARY KEY);"
+ "SELECT 2/0;" // Runtime error.
));
Iterator<AsyncSqlCursor<List<Object>>> curItr =
cursors.iterator();
validateSingleResult(curItr.next(), true);
```
already not ?
Also i see confusing exception message :
`org.apache.ignite.sql.SqlException: IGN-SQL-9
TraceId:337957f7-12f0-4056-9b58-cfead6579fac / by zero`
do we need to log additional issue for it ?
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItSqlMultiStatementTest.java:
##########
@@ -117,13 +108,16 @@ void queryWithDynamicParameters() {
+ "INSERT INTO test VALUES(?, DEFAULT);"
+ "INSERT INTO test VALUES(?, ?);";
- fetchAllCursors(runScript(sql, null, 0, "1", 2, 4, "5"));
+ List<AsyncSqlCursor<List<Object>>> cursors =
Review Comment:
what is the endpoint for using SqlQueryProcessor#queryScriptAsync ? is it
jdbc only ? if it so - such a test is redundant i suppose. PreparedStatement
not support multiple statements.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItSqlMultiStatementTxTest.java:
##########
@@ -0,0 +1,301 @@
+/*
+ * 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.sql.engine;
+
+import static
org.apache.ignite.internal.sql.engine.util.SqlTestUtils.assertThrowsSqlException;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.await;
+import static
org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.apache.ignite.lang.ErrorGroups.Sql.RUNTIME_ERR;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasSize;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrowsExactly;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+import java.util.List;
+import org.apache.ignite.internal.sql.BaseSqlMultiStatementTest;
+import org.apache.ignite.internal.sql.engine.util.Commons;
+import org.apache.ignite.internal.tx.InternalTransaction;
+import org.apache.ignite.internal.util.AsyncCursor.BatchedResult;
+import org.apache.ignite.sql.ExternalTransactionNotSupportedException;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests to verify the execution of queries with transaction control
statements.
+ *
+ * @see SqlQueryType#TX_CONTROL
+ */
+@SuppressWarnings("ThrowableNotThrown")
+public class ItSqlMultiStatementTxTest extends BaseSqlMultiStatementTest {
+ /** Default number of rows in the big table. */
+ private static final int BIG_TABLE_ROWS_COUNT = Commons.IN_BUFFER_SIZE * 6;
+
+ /** Number of completed transactions before the test started. */
+ private int numberOfFinishedTransactionsOnStart;
+
+ @BeforeAll
+ void createTable() {
+ sql("CREATE TABLE test (id INT PRIMARY KEY)");
+
+ createTable("big", 1, 1);
+ sql("INSERT INTO big (id) SELECT x FROM TABLE(SYSTEM_RANGE(1, " +
BIG_TABLE_ROWS_COUNT + "));");
+ }
+
+ @AfterEach
+ void clearTestTable() {
+ sql("DELETE FROM test");
Review Comment:
table will be populated on startup: @BeforeAll and cleared after first test,
is it ok ?
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItSqlMultiStatementTxTest.java:
##########
@@ -0,0 +1,301 @@
+/*
+ * 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.sql.engine;
+
+import static
org.apache.ignite.internal.sql.engine.util.SqlTestUtils.assertThrowsSqlException;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.await;
+import static
org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.apache.ignite.lang.ErrorGroups.Sql.RUNTIME_ERR;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasSize;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrowsExactly;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+import java.util.List;
+import org.apache.ignite.internal.sql.BaseSqlMultiStatementTest;
+import org.apache.ignite.internal.sql.engine.util.Commons;
+import org.apache.ignite.internal.tx.InternalTransaction;
+import org.apache.ignite.internal.util.AsyncCursor.BatchedResult;
+import org.apache.ignite.sql.ExternalTransactionNotSupportedException;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests to verify the execution of queries with transaction control
statements.
+ *
+ * @see SqlQueryType#TX_CONTROL
+ */
+@SuppressWarnings("ThrowableNotThrown")
+public class ItSqlMultiStatementTxTest extends BaseSqlMultiStatementTest {
+ /** Default number of rows in the big table. */
+ private static final int BIG_TABLE_ROWS_COUNT = Commons.IN_BUFFER_SIZE * 6;
+
+ /** Number of completed transactions before the test started. */
+ private int numberOfFinishedTransactionsOnStart;
+
+ @BeforeAll
+ void createTable() {
+ sql("CREATE TABLE test (id INT PRIMARY KEY)");
+
+ createTable("big", 1, 1);
+ sql("INSERT INTO big (id) SELECT x FROM TABLE(SYSTEM_RANGE(1, " +
BIG_TABLE_ROWS_COUNT + "));");
+ }
+
+ @AfterEach
+ void clearTestTable() {
+ sql("DELETE FROM test");
+ }
+
+ @BeforeEach
+ void saveFinishedTxCount() {
+ numberOfFinishedTransactionsOnStart = txManager().finished();
+ }
+
+ @Test
+ void emptyTransactionControlStatement() {
+ List<AsyncSqlCursor<List<Object>>> cursors = fetchAllCursors(
+ runScript("START TRANSACTION; COMMIT"));
+
+ assertThat(cursors, hasSize(2));
+
+ validateSingleResult(cursors.get(0));
+ validateSingleResult(cursors.get(1));
Review Comment:
```suggestion
AsyncSqlCursor<List<Object>> cur = cursors.get(0);
assertTrue(cur.hasNextResult());
validateSingleResult(cur);
cur = cursors.get(1);
assertFalse(cur.hasNextResult());
validateSingleResult(cur);
```
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItSqlMultiStatementTest.java:
##########
@@ -24,86 +24,77 @@
import static org.apache.ignite.lang.ErrorGroups.Sql.RUNTIME_ERR;
import static org.apache.ignite.lang.ErrorGroups.Sql.STMT_VALIDATION_ERR;
import static
org.apache.ignite.lang.ErrorGroups.Transactions.TX_FAILED_READ_WRITE_OPERATION_ERR;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.nullValue;
-import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
-import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.concurrent.CompletableFuture;
-import org.apache.ignite.internal.sql.BaseSqlIntegrationTest;
-import org.apache.ignite.internal.sql.engine.property.SqlPropertiesHelper;
+import org.apache.ignite.internal.sql.BaseSqlMultiStatementTest;
import org.apache.ignite.internal.tx.InternalTransaction;
-import org.apache.ignite.internal.util.AsyncCursor.BatchedResult;
import org.apache.ignite.tx.Transaction;
-import org.jetbrains.annotations.Nullable;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
/**
* Tests to verify the execution of queries with multiple statements.
*/
@SuppressWarnings("ThrowableNotThrown")
-public class ItSqlMultiStatementTest extends BaseSqlIntegrationTest {
+public class ItSqlMultiStatementTest extends BaseSqlMultiStatementTest {
@AfterEach
void dropTables() {
dropAllTables();
}
- @AfterEach
- void checkNoPendingTransactions() {
- assertEquals(0, txManager().pending());
- }
-
@Test
void basicMultiStatementQuery() {
String sql = "CREATE TABLE test (id INT PRIMARY KEY, val INT);"
+ "INSERT INTO test VALUES (0, 0);"
+ + "UPDATE test SET val=1 where id=0;"
+ "EXPLAIN PLAN FOR SELECT * FROM test;"
- + "SELECT * FROM test";
+ + "SELECT * FROM test;"
+ + "DELETE FROM test";
List<AsyncSqlCursor<List<Object>>> cursors =
fetchAllCursors(runScript(sql));
- assertNotNull(cursors);
-
Iterator<AsyncSqlCursor<List<Object>>> curItr = cursors.iterator();
validateSingleResult(curItr.next(), true);
validateSingleResult(curItr.next(), 1L);
+ validateSingleResult(curItr.next(), 1L);
assertNotNull(curItr.next()); // skip EXPLAIN.
- validateSingleResult(curItr.next(), 0, 0);
+ validateSingleResult(curItr.next(), 0, 1);
+ validateSingleResult(curItr.next(), 1L);
assertFalse(curItr.hasNext());
+ cursors.forEach(AsyncSqlCursor::closeAsync);
+
// Ensures that the script is executed completely, even if the cursor
data has not been read.
Review Comment:
you call fetchAllCursors thus as i can see - you completely read all data.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItSqlMultiStatementTxTest.java:
##########
@@ -0,0 +1,301 @@
+/*
+ * 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.sql.engine;
+
+import static
org.apache.ignite.internal.sql.engine.util.SqlTestUtils.assertThrowsSqlException;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.await;
+import static
org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.apache.ignite.lang.ErrorGroups.Sql.RUNTIME_ERR;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasSize;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrowsExactly;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+import java.util.List;
+import org.apache.ignite.internal.sql.BaseSqlMultiStatementTest;
+import org.apache.ignite.internal.sql.engine.util.Commons;
+import org.apache.ignite.internal.tx.InternalTransaction;
+import org.apache.ignite.internal.util.AsyncCursor.BatchedResult;
+import org.apache.ignite.sql.ExternalTransactionNotSupportedException;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests to verify the execution of queries with transaction control
statements.
+ *
+ * @see SqlQueryType#TX_CONTROL
+ */
+@SuppressWarnings("ThrowableNotThrown")
+public class ItSqlMultiStatementTxTest extends BaseSqlMultiStatementTest {
+ /** Default number of rows in the big table. */
+ private static final int BIG_TABLE_ROWS_COUNT = Commons.IN_BUFFER_SIZE * 6;
+
+ /** Number of completed transactions before the test started. */
+ private int numberOfFinishedTransactionsOnStart;
+
+ @BeforeAll
+ void createTable() {
+ sql("CREATE TABLE test (id INT PRIMARY KEY)");
+
+ createTable("big", 1, 1);
+ sql("INSERT INTO big (id) SELECT x FROM TABLE(SYSTEM_RANGE(1, " +
BIG_TABLE_ROWS_COUNT + "));");
+ }
+
+ @AfterEach
+ void clearTestTable() {
+ sql("DELETE FROM test");
+ }
+
+ @BeforeEach
+ void saveFinishedTxCount() {
+ numberOfFinishedTransactionsOnStart = txManager().finished();
+ }
+
+ @Test
+ void emptyTransactionControlStatement() {
+ List<AsyncSqlCursor<List<Object>>> cursors = fetchAllCursors(
+ runScript("START TRANSACTION; COMMIT"));
+
+ assertThat(cursors, hasSize(2));
+
+ validateSingleResult(cursors.get(0));
+ validateSingleResult(cursors.get(1));
+
+ verifyFinishedTxCount(1);
+ }
+
+ @Test
+ void basicTxStatements() {
+ List<AsyncSqlCursor<List<Object>>> cursors = fetchAllCursors(runScript(
+ "START TRANSACTION;"
+ + "INSERT INTO test VALUES(0);"
+ + "INSERT INTO test VALUES(1);"
+ + "COMMIT;"
+
+ + "START TRANSACTION;"
+ + "COMMIT;"
+
+ + "START TRANSACTION;"
+ + "INSERT INTO test VALUES(2);"
+ + "COMMIT;"
+ ));
+
+ verifyFinishedTxCount(3);
+
+ assertQuery("select count(id) from test")
+ .returns(3L).check();
+
+ cursors.forEach(AsyncSqlCursor::closeAsync);
+ }
+
+ @Test
+ void readMoreRowsThanCanBePrefetchedReadOnlyTx() {
+ String query = "START TRANSACTION READ ONLY;"
+ + "SELECT 1;"
+ + "SELECT id FROM big;"
+ + "COMMIT;";
+
+ List<AsyncSqlCursor<List<Object>>> cursors =
fetchAllCursors(runScript(query));
+ assertThat(cursors, hasSize(4));
+
+ BatchedResult<List<Object>> res =
await(cursors.get(2).requestNextAsync(BIG_TABLE_ROWS_COUNT * 2));
+ assertNotNull(res);
+ assertThat(res.items(), hasSize(BIG_TABLE_ROWS_COUNT));
+
+ verifyFinishedTxCount(1);
+
+ cursors.forEach(AsyncSqlCursor::closeAsync);
+ }
+
+ @Test
+ void readMoreRowsThanCanBePrefetchedReadWriteTx() {
+ String query = "START TRANSACTION;"
+ + "UPDATE big SET salary=1 WHERE id=1;"
+ + "SELECT id FROM big;"
+ + "COMMIT;";
+
+ AsyncSqlCursor<List<Object>> cursor = runScript(query);
+
+ assertTrue(cursor.hasNextResult());
+
+ AsyncSqlCursor<List<Object>> updateCursor = await(cursor.nextResult());
+ assertNotNull(updateCursor);
+ validateSingleResult(updateCursor, 1L);
+ assertTrue(updateCursor.hasNextResult());
+
+ AsyncSqlCursor<List<Object>> selectCursor =
await(updateCursor.nextResult());
+ assertNotNull(selectCursor);
+
+ BatchedResult<List<Object>> res = await(
+ selectCursor.requestNextAsync(BIG_TABLE_ROWS_COUNT / 2));
+
+ assertNotNull(res);
+ assertThat(res.items(), hasSize(BIG_TABLE_ROWS_COUNT / 2));
+ assertEquals(1, txManager().pending(), "Transaction must not finished
until the cursor is closed.");
+ assertFalse(selectCursor.nextResult().isDone());
+
+ await(selectCursor.requestNextAsync(BIG_TABLE_ROWS_COUNT)); // Cursor
must close implicitly.
+ verifyFinishedTxCount(1);
+ }
+
+ @Test
+ void openedScriptTransactionRollsBackImplicitly() {
+ {
+ fetchAllCursors(runScript("START TRANSACTION;"));
Review Comment:
seems we need to verify validateSingleResult for single "START TRANSACTION;"
and "COMMIT;" statements here and compare it with referral data bases, wdyt ?
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItSqlMultiStatementTxTest.java:
##########
@@ -0,0 +1,301 @@
+/*
+ * 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.sql.engine;
+
+import static
org.apache.ignite.internal.sql.engine.util.SqlTestUtils.assertThrowsSqlException;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.await;
+import static
org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.apache.ignite.lang.ErrorGroups.Sql.RUNTIME_ERR;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasSize;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrowsExactly;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+import java.util.List;
+import org.apache.ignite.internal.sql.BaseSqlMultiStatementTest;
+import org.apache.ignite.internal.sql.engine.util.Commons;
+import org.apache.ignite.internal.tx.InternalTransaction;
+import org.apache.ignite.internal.util.AsyncCursor.BatchedResult;
+import org.apache.ignite.sql.ExternalTransactionNotSupportedException;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests to verify the execution of queries with transaction control
statements.
+ *
+ * @see SqlQueryType#TX_CONTROL
+ */
+@SuppressWarnings("ThrowableNotThrown")
+public class ItSqlMultiStatementTxTest extends BaseSqlMultiStatementTest {
+ /** Default number of rows in the big table. */
+ private static final int BIG_TABLE_ROWS_COUNT = Commons.IN_BUFFER_SIZE * 6;
+
+ /** Number of completed transactions before the test started. */
+ private int numberOfFinishedTransactionsOnStart;
+
+ @BeforeAll
+ void createTable() {
+ sql("CREATE TABLE test (id INT PRIMARY KEY)");
+
+ createTable("big", 1, 1);
+ sql("INSERT INTO big (id) SELECT x FROM TABLE(SYSTEM_RANGE(1, " +
BIG_TABLE_ROWS_COUNT + "));");
+ }
+
+ @AfterEach
+ void clearTestTable() {
+ sql("DELETE FROM test");
+ }
+
+ @BeforeEach
+ void saveFinishedTxCount() {
+ numberOfFinishedTransactionsOnStart = txManager().finished();
+ }
+
+ @Test
+ void emptyTransactionControlStatement() {
+ List<AsyncSqlCursor<List<Object>>> cursors = fetchAllCursors(
+ runScript("START TRANSACTION; COMMIT"));
+
+ assertThat(cursors, hasSize(2));
+
+ validateSingleResult(cursors.get(0));
+ validateSingleResult(cursors.get(1));
+
+ verifyFinishedTxCount(1);
+ }
+
+ @Test
+ void basicTxStatements() {
+ List<AsyncSqlCursor<List<Object>>> cursors = fetchAllCursors(runScript(
+ "START TRANSACTION;"
+ + "INSERT INTO test VALUES(0);"
+ + "INSERT INTO test VALUES(1);"
+ + "COMMIT;"
+
+ + "START TRANSACTION;"
+ + "COMMIT;"
+
+ + "START TRANSACTION;"
+ + "INSERT INTO test VALUES(2);"
+ + "COMMIT;"
+ ));
+
+ verifyFinishedTxCount(3);
+
+ assertQuery("select count(id) from test")
+ .returns(3L).check();
+
+ cursors.forEach(AsyncSqlCursor::closeAsync);
+ }
+
+ @Test
+ void readMoreRowsThanCanBePrefetchedReadOnlyTx() {
+ String query = "START TRANSACTION READ ONLY;"
+ + "SELECT 1;"
+ + "SELECT id FROM big;"
+ + "COMMIT;";
+
+ List<AsyncSqlCursor<List<Object>>> cursors =
fetchAllCursors(runScript(query));
+ assertThat(cursors, hasSize(4));
+
+ BatchedResult<List<Object>> res =
await(cursors.get(2).requestNextAsync(BIG_TABLE_ROWS_COUNT * 2));
+ assertNotNull(res);
+ assertThat(res.items(), hasSize(BIG_TABLE_ROWS_COUNT));
+
+ verifyFinishedTxCount(1);
+
+ cursors.forEach(AsyncSqlCursor::closeAsync);
+ }
+
+ @Test
+ void readMoreRowsThanCanBePrefetchedReadWriteTx() {
+ String query = "START TRANSACTION;"
+ + "UPDATE big SET salary=1 WHERE id=1;"
+ + "SELECT id FROM big;"
+ + "COMMIT;";
+
+ AsyncSqlCursor<List<Object>> cursor = runScript(query);
+
+ assertTrue(cursor.hasNextResult());
+
+ AsyncSqlCursor<List<Object>> updateCursor = await(cursor.nextResult());
+ assertNotNull(updateCursor);
+ validateSingleResult(updateCursor, 1L);
+ assertTrue(updateCursor.hasNextResult());
+
+ AsyncSqlCursor<List<Object>> selectCursor =
await(updateCursor.nextResult());
+ assertNotNull(selectCursor);
+
+ BatchedResult<List<Object>> res = await(
+ selectCursor.requestNextAsync(BIG_TABLE_ROWS_COUNT / 2));
+
+ assertNotNull(res);
+ assertThat(res.items(), hasSize(BIG_TABLE_ROWS_COUNT / 2));
+ assertEquals(1, txManager().pending(), "Transaction must not finished
until the cursor is closed.");
+ assertFalse(selectCursor.nextResult().isDone());
+
+ await(selectCursor.requestNextAsync(BIG_TABLE_ROWS_COUNT)); // Cursor
must close implicitly.
+ verifyFinishedTxCount(1);
+ }
+
+ @Test
+ void openedScriptTransactionRollsBackImplicitly() {
+ {
+ fetchAllCursors(runScript("START TRANSACTION;"));
+ verifyFinishedTxCount(1);
+ }
+
+ {
+ List<AsyncSqlCursor<List<Object>>> cursors = fetchAllCursors(
+ runScript("START TRANSACTION;"
+ + "INSERT INTO test VALUES(0);"
+ + "INSERT INTO test VALUES(1);"
+ + "COMMIT;"
+
+ + "START TRANSACTION;"
+ + "INSERT INTO test VALUES(2);"
+ + "SELECT * FROM test;"
+ )
+ );
+
+ verifyFinishedTxCount(3);
+
+ assertThat(cursors, hasSize(7));
+ cursors.subList(0, 4).forEach(AsyncSqlCursor::closeAsync);
+
+ // Cursors associated with a transaction must be closed when the
transaction is rolled back.
+ checkNoPendingTransactionsAndOpenedCursors();
+
+ assertQuery("select count(id) from test")
+ .returns(2L).check();
Review Comment:
lets additionally check :
```
cursors.subList(5, 7).forEach(cur -> await(cur.closeAsync()));
assertQuery("select count(id) from test")
.returns(2L).check();
```
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/tx/ScriptTransactionHandler.java:
##########
@@ -0,0 +1,319 @@
+/*
+ * 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.sql.engine.tx;
+
+import static org.apache.ignite.lang.ErrorGroups.Sql.RUNTIME_ERR;
+
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.ignite.internal.sql.engine.SqlQueryType;
+import org.apache.ignite.internal.sql.engine.sql.IgniteSqlCommitTransaction;
+import org.apache.ignite.internal.sql.engine.sql.IgniteSqlStartTransaction;
+import org.apache.ignite.internal.sql.engine.sql.IgniteSqlStartTransactionMode;
+import org.apache.ignite.internal.sql.engine.sql.ParsedResult;
+import org.apache.ignite.internal.sql.engine.util.Commons;
+import org.apache.ignite.internal.tx.InternalTransaction;
+import org.apache.ignite.internal.util.AsyncCursor;
+import org.apache.ignite.sql.ExternalTransactionNotSupportedException;
+import org.apache.ignite.sql.SqlException;
+import org.apache.ignite.tx.IgniteTransactions;
+import org.apache.ignite.tx.TransactionOptions;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Starts an implicit transaction if there is no external transaction.
Supports script transaction management using
+ * {@link SqlQueryType#TX_CONTROL} statements.
+ */
+public class ScriptTransactionHandler extends QueryTransactionHandler {
+ /** No-op transaction wrapper. */
+ private static final QueryTransactionWrapper NOOP_TX_WRAPPER = new
NoopTransactionWrapper();
+
+ /** Wraps a transaction, which is managed by SQL engine via {@link
SqlQueryType#TX_CONTROL} statements. */
+ private volatile @Nullable ManagedTransactionWrapper wrapper;
+
+ public ScriptTransactionHandler(IgniteTransactions transactions, @Nullable
InternalTransaction externalTransaction) {
+ super(transactions, externalTransaction);
+ }
+
+ @Override
+ protected @Nullable InternalTransaction activeTransaction() {
+ return externalTransaction == null ? scriptTransaction() :
externalTransaction;
+ }
+
+ /**
+ * Starts a transaction if there is no external transaction.
+ *
+ * @param parsedResult Parse result.
+ * @param cursorFut Cursor future for the current statement.
+ * @return Transaction wrapper.
+ */
+ public QueryTransactionWrapper startScriptTxIfNeeded(
+ ParsedResult parsedResult,
+ CompletableFuture<? extends AsyncCursor<?>> cursorFut
+ ) {
+ try {
+ SqlQueryType queryType = parsedResult.queryType();
+
+ if (queryType == SqlQueryType.TX_CONTROL) {
+ if (externalTransaction != null) {
+ throw new ExternalTransactionNotSupportedException();
+ }
+
+ return handleTxControlStatement(parsedResult.parsedTree());
+ }
+
+ ManagedTransactionWrapper wrapper = this.wrapper;
+
+ if (wrapper == null) {
+ return startTxIfNeeded(queryType);
+ }
+
+ validateStatement(parsedResult.queryType(), wrapper.unwrap());
+
+ return wrapper.forStatement(queryType, cursorFut);
+ } catch (SqlException e) {
+ InternalTransaction scriptTx = scriptTransaction();
+
+ if (scriptTx != null) {
+ scriptTx.rollback();
+ }
+
+ throw e;
+ }
+ }
+
+ private @Nullable InternalTransaction scriptTransaction() {
+ ManagedTransactionWrapper hld = wrapper;
+
+ return hld != null ? hld.unwrap() : null;
+ }
+
+ private QueryTransactionWrapper handleTxControlStatement(SqlNode node) {
+ ManagedTransactionWrapper txWrapper = this.wrapper;
+
+ if (node instanceof IgniteSqlCommitTransaction) {
+ if (txWrapper == null) {
+ return NOOP_TX_WRAPPER;
+ }
+
+ this.wrapper = null;
+
+ return txWrapper.forCommit();
Review Comment:
looking at this place i expected test will pass, but it didn\`t, i\`m wrong
? i.e. i see that RO transaction already commited but we still have partially
readed cursors.
```
@Test
void readMoreRowsThanCanBePrefetchedReadOnlyTx() {
String query = "START TRANSACTION READ ONLY;"
+ "SELECT id FROM big;"
+ "SELECT 1;"
+ "COMMIT;";
List<AsyncSqlCursor<List<Object>>> cursors =
fetchAllCursors(runScript(query));
assertThat(cursors, hasSize(4));
verifyFinishedTxCount(0);
BatchedResult<List<Object>> res =
await(cursors.get(2).requestNextAsync(1));
assertNotNull(res);
verifyFinishedTxCount(0);
res = await(cursors.get(2).requestNextAsync(100));
assertNotNull(res);
verifyFinishedTxCount(0);
res = await(cursors.get(1).requestNextAsync(1));
assertNotNull(res);
verifyFinishedTxCount(0);
cursors.forEach(AsyncSqlCursor::closeAsync);
}
```
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItSqlMultiStatementTxTest.java:
##########
@@ -0,0 +1,301 @@
+/*
+ * 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.sql.engine;
+
+import static
org.apache.ignite.internal.sql.engine.util.SqlTestUtils.assertThrowsSqlException;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.await;
+import static
org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.apache.ignite.lang.ErrorGroups.Sql.RUNTIME_ERR;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasSize;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrowsExactly;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+import java.util.List;
+import org.apache.ignite.internal.sql.BaseSqlMultiStatementTest;
+import org.apache.ignite.internal.sql.engine.util.Commons;
+import org.apache.ignite.internal.tx.InternalTransaction;
+import org.apache.ignite.internal.util.AsyncCursor.BatchedResult;
+import org.apache.ignite.sql.ExternalTransactionNotSupportedException;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests to verify the execution of queries with transaction control
statements.
+ *
+ * @see SqlQueryType#TX_CONTROL
+ */
+@SuppressWarnings("ThrowableNotThrown")
+public class ItSqlMultiStatementTxTest extends BaseSqlMultiStatementTest {
+ /** Default number of rows in the big table. */
+ private static final int BIG_TABLE_ROWS_COUNT = Commons.IN_BUFFER_SIZE * 6;
+
+ /** Number of completed transactions before the test started. */
+ private int numberOfFinishedTransactionsOnStart;
+
+ @BeforeAll
+ void createTable() {
+ sql("CREATE TABLE test (id INT PRIMARY KEY)");
+
+ createTable("big", 1, 1);
+ sql("INSERT INTO big (id) SELECT x FROM TABLE(SYSTEM_RANGE(1, " +
BIG_TABLE_ROWS_COUNT + "));");
+ }
+
+ @AfterEach
+ void clearTestTable() {
+ sql("DELETE FROM test");
+ }
+
+ @BeforeEach
+ void saveFinishedTxCount() {
+ numberOfFinishedTransactionsOnStart = txManager().finished();
+ }
+
+ @Test
+ void emptyTransactionControlStatement() {
+ List<AsyncSqlCursor<List<Object>>> cursors = fetchAllCursors(
+ runScript("START TRANSACTION; COMMIT"));
+
+ assertThat(cursors, hasSize(2));
+
+ validateSingleResult(cursors.get(0));
+ validateSingleResult(cursors.get(1));
+
+ verifyFinishedTxCount(1);
+ }
+
+ @Test
+ void basicTxStatements() {
+ List<AsyncSqlCursor<List<Object>>> cursors = fetchAllCursors(runScript(
+ "START TRANSACTION;"
+ + "INSERT INTO test VALUES(0);"
+ + "INSERT INTO test VALUES(1);"
+ + "COMMIT;"
+
+ + "START TRANSACTION;"
+ + "COMMIT;"
+
+ + "START TRANSACTION;"
+ + "INSERT INTO test VALUES(2);"
+ + "COMMIT;"
+ ));
+
+ verifyFinishedTxCount(3);
+
+ assertQuery("select count(id) from test")
+ .returns(3L).check();
+
+ cursors.forEach(AsyncSqlCursor::closeAsync);
+ }
+
+ @Test
+ void readMoreRowsThanCanBePrefetchedReadOnlyTx() {
+ String query = "START TRANSACTION READ ONLY;"
+ + "SELECT 1;"
+ + "SELECT id FROM big;"
+ + "COMMIT;";
+
+ List<AsyncSqlCursor<List<Object>>> cursors =
fetchAllCursors(runScript(query));
+ assertThat(cursors, hasSize(4));
+
+ BatchedResult<List<Object>> res =
await(cursors.get(2).requestNextAsync(BIG_TABLE_ROWS_COUNT * 2));
+ assertNotNull(res);
+ assertThat(res.items(), hasSize(BIG_TABLE_ROWS_COUNT));
+
+ verifyFinishedTxCount(1);
+
+ cursors.forEach(AsyncSqlCursor::closeAsync);
+ }
+
+ @Test
+ void readMoreRowsThanCanBePrefetchedReadWriteTx() {
+ String query = "START TRANSACTION;"
+ + "UPDATE big SET salary=1 WHERE id=1;"
+ + "SELECT id FROM big;"
+ + "COMMIT;";
+
+ AsyncSqlCursor<List<Object>> cursor = runScript(query);
+
+ assertTrue(cursor.hasNextResult());
+
+ AsyncSqlCursor<List<Object>> updateCursor = await(cursor.nextResult());
+ assertNotNull(updateCursor);
+ validateSingleResult(updateCursor, 1L);
+ assertTrue(updateCursor.hasNextResult());
+
+ AsyncSqlCursor<List<Object>> selectCursor =
await(updateCursor.nextResult());
+ assertNotNull(selectCursor);
+
+ BatchedResult<List<Object>> res = await(
+ selectCursor.requestNextAsync(BIG_TABLE_ROWS_COUNT / 2));
+
+ assertNotNull(res);
+ assertThat(res.items(), hasSize(BIG_TABLE_ROWS_COUNT / 2));
+ assertEquals(1, txManager().pending(), "Transaction must not finished
until the cursor is closed.");
+ assertFalse(selectCursor.nextResult().isDone());
+
+ await(selectCursor.requestNextAsync(BIG_TABLE_ROWS_COUNT)); // Cursor
must close implicitly.
+ verifyFinishedTxCount(1);
+ }
+
+ @Test
+ void openedScriptTransactionRollsBackImplicitly() {
+ {
+ fetchAllCursors(runScript("START TRANSACTION;"));
+ verifyFinishedTxCount(1);
+ }
+
+ {
+ List<AsyncSqlCursor<List<Object>>> cursors = fetchAllCursors(
+ runScript("START TRANSACTION;"
+ + "INSERT INTO test VALUES(0);"
+ + "INSERT INTO test VALUES(1);"
+ + "COMMIT;"
+
+ + "START TRANSACTION;"
+ + "INSERT INTO test VALUES(2);"
+ + "SELECT * FROM test;"
+ )
+ );
+
+ verifyFinishedTxCount(3);
+
+ assertThat(cursors, hasSize(7));
+ cursors.subList(0, 4).forEach(AsyncSqlCursor::closeAsync);
Review Comment:
```suggestion
cursors.subList(0, 4).forEach(cur -> await(cur.closeAsync()));
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
