zstan commented on code in PR #13366:
URL: https://github.com/apache/ignite/pull/13366#discussion_r3728218617
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExecutionServiceImpl.java:
##########
@@ -581,6 +598,290 @@ 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 deadline = waitMs > 0
+ ? U.currentTimeMillis() + waitMs
+ : waitMs < 0 ? U.currentTimeMillis() : 0L;
+
+ RootQuery<Row> selectQry = qry;
+
+ while (true) {
+ FieldsQueryCursor<List<?>> cursor = tryExecuteForUpdate(selectQry,
plan, userTx, waitMs, deadline);
Review Comment:
i already wrote upper: let`\s simplify code a bit ? I see that "deadline"
only helpful in this function if "waitMs > 0" - thus i think you can pass only
"waitMs" here.
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExecutionServiceImpl.java:
##########
@@ -581,6 +598,290 @@ 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 deadline = waitMs > 0
+ ? U.currentTimeMillis() + waitMs
+ : waitMs < 0 ? U.currentTimeMillis() : 0L;
+
+ RootQuery<Row> selectQry = qry;
+
+ while (true) {
+ FieldsQueryCursor<List<?>> cursor = tryExecuteForUpdate(selectQry,
plan, userTx, waitMs, deadline);
+
+ if (cursor != null)
+ return cursor;
+
+ if (deadline != 0 && U.currentTimeMillis() >= deadline) {
+ throw new IgniteSQLException(
+ IgniteResource.INSTANCE.selectForUpdateLockFailed().str(),
+ IgniteQueryErrorCode.CONCURRENT_UPDATE);
+ }
+
+ // The previous query has already been closed after
materialisation, so retry with a fresh root query.
+ selectQry = qry.retryQuery();
Review Comment:
misunderstand here - if we already obtain materialisation = results ? then -
cursor will return a bit upper. Seems here you need to write smth. like : "in
case of failed lock acquisition ... " and seems all we need here is just to
reset RootNode.state ? Use a constructor is ok but let`\s rename method like:
selectQry = qry.clone(); ?
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/SelectForUpdatePlan.java:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.prepare;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import org.apache.ignite.cache.CacheEntry;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Query plan for {@code SELECT ... FOR UPDATE} statements.
+ *
+ * <p>Wraps an inner {@link MultiStepQueryPlan} whose SELECT list has {@code
_KEY}, {@code _VAL},
+ * and {@code _VER} columns of the tables participating in the query appended
at the end. At execution time
+ * the executor:
+ * <ol>
+ * <li>Runs the inner plan and materialises the full result set.</li>
+ * <li>Extracts the hidden columns described by {@link #lockTargets} from
every row.</li>
+ * <li>Creates {@link CacheEntry} to lock in the current transaction.</li>
+ * <li>Calls {@code cache.lockTxEntries(entries, waitMs)} to acquire
pessimistic locks.</li>
+ * <li>Repeats the inner plan when a selected entry version changes before
locking.</li>
+ * <li>Returns only the first {@link #userColumnCount} columns to the
caller.</li>
+ * </ol>
+ */
+public class SelectForUpdatePlan extends AbstractQueryPlan {
+ /** Inner plan: SELECT with _KEY, _VAL, and _VER columns appended at the
end. */
+ private final MultiStepQueryPlan innerPlan;
+
+ /** Number of user-visible result columns (total inner columns minus the
appended _KEY, _VAL, and _VER). */
+ private final int userColumnCount;
+
+ /**
+ * Lock-wait limit from the {@code FOR UPDATE} clause:
+ * {@code null} = use transaction timeout, {@code 0L} = NOWAIT,
+ * positive value = WAIT n seconds.
+ */
+ @Nullable private final Long waitSeconds;
+
+ /** Tables whose rows must be locked and positions of their hidden columns
in the inner result. */
+ private final List<LockTarget> lockTargets;
+
+ /** */
+ public SelectForUpdatePlan(
+ MultiStepQueryPlan innerPlan,
+ int userColumnCount,
+ @Nullable Long waitSeconds,
+ List<LockTarget> lockTargets
+ ) {
+ super(innerPlan.query());
+
+ this.innerPlan = innerPlan;
+ this.userColumnCount = userColumnCount;
+ this.waitSeconds = waitSeconds;
+ this.lockTargets = Collections.unmodifiableList(new
ArrayList<>(lockTargets));
Review Comment:
```suggestion
this.lockTargets = List.copyOf(lockTargets);
```
--
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]
