zstan commented on code in PR #13366:
URL: https://github.com/apache/ignite/pull/13366#discussion_r3749880318
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExecutionServiceImpl.java:
##########
@@ -581,6 +601,388 @@ private FieldsQueryCursor<List<?>>
executeDdl(RootQuery<Row> qry, DdlPlan plan)
}
}
+ /**
+ * Executes a {@code SELECT ... FOR UPDATE} plan.
+ *
+ * <ol>
+ * <li>Validates that the current transaction is PESSIMISTIC.</li>
+ * <li>Runs the inner SELECT with hidden key, value, and version columns
and materialises all rows.</li>
+ * <li>Builds cache entries from the hidden columns.</li>
+ * <li>Creates a savepoint, acquires pessimistic locks via {@code
lockTxEntries()},
+ * and releases the savepoint on success (or rolls back on
failure).</li>
+ * <li>Repeats the SELECT and lock attempt after a concurrent version
change while the deadline permits.</li>
+ * <li>Returns a cursor with only the user-visible columns (the appended
_KEY is stripped).</li>
+ * </ol>
+ */
+ private FieldsQueryCursor<List<?>> executeForUpdate(RootQuery<Row> qry,
SelectForUpdatePlan plan) {
+ GridNearTxLocal userTx = Commons.queryTransaction(qry.context(),
ctx.cache().context());
+
+ if (userTx == null || !userTx.pessimistic())
+ throw new IgniteSQLException(
+
IgniteResource.INSTANCE.selectForUpdateRequiresPessimisticTx().str(),
+ IgniteQueryErrorCode.UNSUPPORTED_OPERATION);
+
+ long waitMs = waitMillis(plan);
+
+ // Zero means that retries are limited only by the transaction or
query timeout.
+ long lockAcquisitionEndTime = waitMs > 0
+ ? U.currentTimeMillis() + waitMs
+ : waitMs < 0 ? U.currentTimeMillis() : 0L;
+
+ RootQuery<Row> selectQry = qry;
+
+ while (true) {
+ FieldsQueryCursor<List<?>> cursor = tryExecuteForUpdate(selectQry,
plan, userTx, waitMs, lockAcquisitionEndTime);
+
+ if (cursor != null)
+ return cursor;
+
+ if (lockAcquisitionEndTime != 0 && U.currentTimeMillis() >=
lockAcquisitionEndTime) {
+ throw new IgniteSQLException(
+ IgniteResource.INSTANCE.selectForUpdateLockFailed().str(),
+ IgniteQueryErrorCode.CONCURRENT_UPDATE);
+ }
+
+ // The previous query has already been closed after execution, so
retry with a fresh root query.
+ selectQry = qry.retryQuery();
+ qryReg.register(selectQry);
+ }
+ }
+
+ /**
+ * Converts the SQL lock wait value to the internal millisecond
representation.
+ *
+ * @param plan SELECT FOR UPDATE plan.
+ * @return {@code 0} for the remaining transaction/query timeout, {@code
-1} for NOWAIT,
+ * or a positive timeout in milliseconds.
+ */
+ private static long waitMillis(SelectForUpdatePlan plan) {
+ // Convert SQL waitSeconds to the internal lock-wait representation:
+ // null means use the remaining transaction time or the query timeout
and is encoded as 0;
+ // 0 requests NOWAIT and is encoded as -1; a positive value is
converted from seconds to milliseconds.
+ Long waitSeconds = plan.waitSeconds();
+
+ if (waitSeconds == null)
+ return 0L;
+ else if (waitSeconds == 0L)
+ return -1L;
+ else
+ return waitSeconds * 1000L;
+ }
+
+ /**
+ * Executes the inner SELECT and attempts to acquire transaction locks for
the selected row versions.
+ *
+ * @param qry Root query for this execution attempt.
+ * @param plan SELECT FOR UPDATE plan.
+ * @param userTx Transaction that acquires the locks.
+ * @param waitMs Lock wait time in the internal representation.
+ * @param lockAcquisitionEndTime Absolute lock acquisition deadline in
milliseconds.
+ * @return Result cursor if all required locks were acquired, or {@code
null} if at least one lock was not acquired.
+ */
+ @Nullable private FieldsQueryCursor<List<?>> tryExecuteForUpdate(
+ RootQuery<Row> qry,
+ SelectForUpdatePlan plan,
+ GridNearTxLocal userTx,
+ long waitMs,
+ long lockAcquisitionEndTime
+ ) {
+ // Run the inner SELECT (with _KEY, _VAL, _VER appended) and collect
all rows.
+ ListFieldsQueryCursor<?> innerCursor = mapAndExecutePlan(qry,
plan.innerPlan());
+
+ // TODO: IGNITE-28957 SELECT FOR UPDATE may cause OOM by materializing
the entire result set.
+ List<List<?>> rows = innerCursor.getAll();
+
+ int userColCnt = plan.userColumnCount();
+
+ if (rows.isEmpty())
+ return createResultCursor(qry, plan, rows, userColCnt);
+
+ List<Map.Entry<IgniteInternalCache<Object, Object>, Map<Object,
CacheEntry<Object, Object>>>> lockBatches =
+ collectLockBatches(plan, rows);
+
+ if (!tryAcquireLocks(userTx, lockBatches, waitMs,
lockAcquisitionEndTime))
+ return null;
+
+ return createResultCursor(qry, plan, rows, userColCnt);
+ }
+
+ /**
+ * Collects unique cache entries to lock and orders them by cache ID,
partition ID, key hash, and key bytes.
+ *
+ * @param plan SELECT FOR UPDATE plan containing the lock targets.
+ * @param rows Selected rows containing the internal lock columns.
+ * @return Ordered cache-entry batches.
+ */
+ private List<Map.Entry<IgniteInternalCache<Object, Object>, Map<Object,
CacheEntry<Object, Object>>>>
+ collectLockBatches(SelectForUpdatePlan plan, List<List<?>> rows) {
+ Map<IgniteInternalCache<Object, Object>, Map<Object,
CacheEntry<Object, Object>>> lockBatches =
+ new TreeMap<>(Comparator.comparingInt(cache ->
cache.context().cacheId()));
+
+ for (LockTarget target : plan.lockTargets()) {
+ SchemaPlus schemaPlus = schemaHolder.schema(target.schemaName());
+
+ if (schemaPlus == null)
+ throw new IgniteSQLException("Schema not found: " +
target.schemaName(),
+ IgniteQueryErrorCode.SCHEMA_NOT_FOUND);
+
+ IgniteTable igniteTable =
(IgniteTable)schemaPlus.getTable(target.tableName());
+
+ if (igniteTable == null)
+ throw new IgniteSQLException("Table not found: " +
target.tableName(),
+ IgniteQueryErrorCode.TABLE_NOT_FOUND);
+
+ GridCacheContext<Object, Object> cctx =
+ (GridCacheContext<Object,
Object>)((CacheTableDescriptor)igniteTable.descriptor()).cacheContext();
+
+ IgniteInternalCache<Object, Object> cache =
cctx.cache().keepBinary();
Review Comment:
why do you need "keepBinary()" here ? Seems it work properly well without
##########
modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/SelectForUpdateIntegrationTest.java:
##########
@@ -0,0 +1,682 @@
+/*
+ * 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.processors.query.calcite.integration;
+
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.TransactionConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.TestRecordingCommunicationSpi;
+import org.apache.ignite.internal.processors.query.IgniteSQLException;
+import org.apache.ignite.internal.processors.query.QueryEngine;
+import org.apache.ignite.internal.processors.query.calcite.QueryChecker;
+import
org.apache.ignite.internal.processors.query.calcite.message.QueryBatchMessage;
+import org.apache.ignite.internal.processors.query.calcite.util.Commons;
+import org.apache.ignite.internal.processors.query.calcite.util.IgniteResource;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.transactions.Transaction;
+import org.junit.Test;
+
+import static
org.apache.ignite.internal.processors.query.calcite.integration.AbstractBasicIntegrationTransactionalTest.SqlTransactionMode.ALL;
+import static org.apache.ignite.transactions.TransactionConcurrency.OPTIMISTIC;
+import static
org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
+import static
org.apache.ignite.transactions.TransactionIsolation.READ_COMMITTED;
+import static org.apache.ignite.transactions.TransactionState.ACTIVE;
+
+/**
+ * Integration tests for {@code SELECT ... FOR UPDATE} syntax.
+ */
+public class SelectForUpdateIntegrationTest extends
AbstractBasicIntegrationTest {
+ /** */
+ private static IgniteEx ignite0;
+
+ /** */
+ private static IgniteEx ignite1;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String
igniteInstanceName) throws Exception {
+ return super.getConfiguration(igniteInstanceName)
+ .setTransactionConfiguration(new TransactionConfiguration()
+ .setTxAwareQueriesEnabled(true))
+ .setCommunicationSpi(new TestRecordingCommunicationSpi());
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ ignite0 = grid(0);
+ ignite1 = grid(1);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ ignite0 = null;
+ ignite1 = null;
+
+ super.afterTestsStopped();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected QueryChecker assertQuery(Ignite ignite, String qry) {
+ Transaction tx = ignite.transactions().tx();
+ QueryChecker checker;
+
+ if (tx == null)
+ checker = super.assertQuery(ignite, qry);
+ else {
+ checker = new QueryChecker(qry, tx, ALL) {
+ @Override public void check() {
+ tx.suspend();
+
+ try {
+ super.check();
+ }
+ finally {
+ tx.resume();
+ }
+ }
+
+ @Override protected QueryEngine getEngine() {
+ return
Commons.lookupComponent(((IgniteEx)ignite).context(), QueryEngine.class);
+ }
+ };
+ }
+
+ return checker;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ super.beforeTest();
+
+ sql("CREATE TABLE Person (id INT PRIMARY KEY, name VARCHAR, age INT,
deptId INT, managerId INT) " +
+ "WITH atomicity=TRANSACTIONAL");
+ sql("INSERT INTO Person (id, name, age) VALUES " +
+ "(1, 'Alice', 20), (2, 'Bob', 21), (3, 'Ann', 22), (4, 'Bill',
23)" +
+ ", (5, 'Alex', 24), (6, 'Ben', 25), (7, 'Cathy', 26), (8, 'Carl',
27), (9, 'Diana', 28)" +
+ ", (10, 'David', 29), (11, 'Eva', 30), (12, 'Evan', 31), (13,
'Fiona', 32), (14, 'Frank', 33)" +
+ ", (15, 'Grace', 34), (16, 'George', 35), (17, 'Hannah', 36), (18,
'Harry', 37), (19, 'Ivy', 38)" +
+ ", (20, 'Ian', 39), (21, 'Jack', 40), (22, 'Jill', 41), (23,
'Karen', 42), (24, 'Kyle', 43)" +
+ ", (25, 'Laura', 44), (26, 'Leo', 45), (27, 'Mia', 46), (28,
'Mike', 47), (29, 'Nina', 48)" +
+ ", (30, 'Nick', 49)");
+ sql("UPDATE Person SET deptId = 1, managerId = 2 WHERE id = 1");
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ sql("DROP TABLE IF EXISTS Dept");
+ sql("DROP TABLE IF EXISTS Person");
+
+ super.afterTest();
+ }
+
+ /** SELECT FOR UPDATE without OF locks rows of every table participating
in a JOIN. */
+ @Test
+ public void testSelectForUpdateJoinLocksAllTables() throws Exception {
+ createDeptTable();
+
+ CountDownLatch locked = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+
+ IgniteInternalFuture<?> lockFut = GridTestUtils.runAsync(() -> {
+ try (Transaction tx = ignite0.transactions().txStart(PESSIMISTIC,
READ_COMMITTED)) {
+ assertQuery(ignite0,
+ "SELECT p.id FROM Person p JOIN Dept d ON p.deptId = d.id
WHERE p.id = 1 FOR UPDATE")
Review Comment:
if i change it to NON pk like : p.deptId = d.**name** it will just silently
process to infinitelly execute, is it expected or i miss smth ?
--
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]
