Copilot commented on code in PR #13366:
URL: https://github.com/apache/ignite/pull/13366#discussion_r3747911557
##########
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;
Review Comment:
`waitSeconds * 1000L` can overflow (e.g. for large WAIT values) and turn the
computed lock wait into a negative number, which changes semantics (e.g. can
behave like NOWAIT). Consider saturating or explicitly guarding against
overflow when converting seconds to milliseconds.
##########
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();
+ Map<Object, CacheEntry<Object, Object>> entries =
+ lockBatches.computeIfAbsent(cache, key -> new
LinkedHashMap<>());
+ int keyColumnIdx = target.keyColumnIndex();
+
+ for (List<?> row : rows) {
+ Object key = row.get(keyColumnIdx);
+
+ // An outer join has no row to lock on its non-matching side.
+ if (key == null)
+ continue;
+
+ Object val = row.get(keyColumnIdx + 1);
+ GridCacheVersion ver = (GridCacheVersion)row.get(keyColumnIdx
+ 2);
+
+ // JOINs can repeat a row, but a transaction needs only one
lock per cache key.
+ entries.put(key, new CacheEntryImplEx<>(key, val, ver));
+ }
+ }
+
+ for (Map.Entry<IgniteInternalCache<Object, Object>, Map<Object,
CacheEntry<Object, Object>>> batch :
+ lockBatches.entrySet()) {
+ Map<Object, CacheEntry<Object, Object>> entries = batch.getValue();
+
+ orderLockEntries(batch.getKey().context(), entries);
+ }
+
+ return new ArrayList<>(lockBatches.entrySet());
+ }
+
+ /**
+ * Orders entries by partition, key hash, and serialized key bytes in case
of a hash collision.
+ *
+ * @param cctx Cache context used to prepare cache keys.
+ * @param entries Entries to order.
+ */
+ private static void orderLockEntries(
+ GridCacheContext<Object, Object> cctx,
+ Map<Object, CacheEntry<Object, Object>> entries
+ ) {
+ List<LockEntry> orderedEntries = new ArrayList<>(entries.size());
+
+ for (CacheEntry<Object, Object> entry : entries.values()) {
+ KeyCacheObject key = cctx.toCacheKeyObject(entry.getKey());
+
+ orderedEntries.add(new LockEntry(entry, key));
+ }
+
+ orderedEntries.sort(Comparator.comparingInt((LockEntry entry) ->
entry.part)
+ .thenComparingInt(entry -> entry.keyHash));
+
+ for (int start = 0; start < orderedEntries.size(); ) {
+ LockEntry first = orderedEntries.get(start);
+ int end = start + 1;
+
+ while (end < orderedEntries.size()
+ && orderedEntries.get(end).part == first.part
+ && orderedEntries.get(end).keyHash == first.keyHash)
+ end++;
+
+ if (end - start > 1) {
+ try {
+ for (int i = start; i < end; i++)
+
orderedEntries.get(i).prepareKeyBytes(cctx.cacheObjectContext());
+ }
+ catch (IgniteCheckedException e) {
+ throw new IgniteSQLException("Failed to serialize a cache
key for lock ordering", e);
+ }
+
+ orderedEntries.subList(start, end).sort(Comparator.comparing(
+ entry -> entry.keyBytes,
+ UnsignedBytes.lexicographicalComparator()
+ ));
+ }
+
+ start = end;
+ }
+
+ entries.clear();
+
+ for (LockEntry entry : orderedEntries)
+ entries.put(entry.entry.getKey(), entry.entry);
+ }
+
+ /** Cache entry with the key attributes used for lock ordering. */
+ private static class LockEntry {
+ /** Cache entry. */
+ private final CacheEntry<Object, Object> entry;
+
+ /** Key partition. */
+ private final int part;
+
+ /** Key hash. */
+ private final int keyHash;
+
+ /** Cache key. */
+ private final KeyCacheObject key;
+
+ /** Serialized key bytes, initialized only for hash collisions. */
+ private byte[] keyBytes;
+
+ /**
+ * @param entry Cache entry.
+ * @param key Cache key.
+ */
+ private LockEntry(CacheEntry<Object, Object> entry, KeyCacheObject
key) {
+ this.entry = entry;
+ this.key = key;
+ part = key.partition();
+ keyHash = key.hashCode();
+ }
+
+ /**
+ * Serializes the key for collision resolution.
+ *
+ * @param ctx Cache object context.
+ * @throws IgniteCheckedException If serialization fails.
+ */
+ private void prepareKeyBytes(CacheObjectValueContext ctx) throws
IgniteCheckedException {
+ keyBytes = key.valueBytes(ctx);
+ }
+ }
+
+ /**
+ * Tries to lock all collected entries within a transaction savepoint.
+ *
+ * @param userTx Transaction that acquires the locks.
+ * @param lockBatches Cache entries grouped by cache.
+ * @param waitMs Lock wait time in the internal representation.
+ * @param lockAcquisitionEndTime Absolute lock deadline, or {@code 0} to
use the transaction/query timeout.
+ * @return {@code true} if every lock was acquired.
+ */
+ private static boolean tryAcquireLocks(
+ GridNearTxLocal userTx,
+ List<Map.Entry<IgniteInternalCache<Object, Object>, Map<Object,
CacheEntry<Object, Object>>>> lockBatches,
+ long waitMs,
+ long lockAcquisitionEndTime
+ ) {
+ try {
+ // lockTxEntries() requires the transaction to be bound to the
current thread
+ // (it checks cctx.tm().threadLocalTx()). Resume it here and
suspend afterwards,
+ // following the same pattern as
ModifyNode.invokeInsideTransaction().
+ userTx.resume();
+
+ try {
+ // Create a savepoint so that a failed lock attempt can be
rolled back without aborting the whole tx.
+ String spName = "_for_update_" + UUID.randomUUID();
+
+ userTx.savepoint(spName, false);
+
+ boolean locked = true;
+
+ try {
+ for (Map.Entry<IgniteInternalCache<Object, Object>,
Map<Object, CacheEntry<Object, Object>>> batch :
+ lockBatches) {
+ if (batch.getValue().isEmpty())
+ continue;
+
+ long batchWaitMs = waitMs;
+
+ if (lockAcquisitionEndTime > 0) {
+ batchWaitMs = lockAcquisitionEndTime -
U.currentTimeMillis();
+
+ // Excluse case where
+ if (batchWaitMs == 0)
+ batchWaitMs = -1L;
+ }
Review Comment:
When a lock acquisition deadline is used, `batchWaitMs` can become negative
(not just 0). `IgniteInternalCache.lockTxEntries` documents only `0` (tx
timeout) and `-1` (NOWAIT); passing other negative values is undefined. Also
the in-code comment is incomplete/has a typo.
##########
modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/BulkOperationDeadlockIntegrationTest.java:
##########
@@ -0,0 +1,348 @@
+/*
+ * 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.io.Serializable;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicLongArray;
+import java.util.function.IntFunction;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.configuration.CacheConfiguration;
+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.testframework.GridTestUtils;
+import org.apache.ignite.transactions.Transaction;
+import org.apache.ignite.transactions.TransactionIsolation;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+import static
org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
+import static
org.apache.ignite.transactions.TransactionIsolation.READ_COMMITTED;
+import static
org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ;
+
+/** Tests that concurrent bulk operations do not deadlock. */
+public class BulkOperationDeadlockIntegrationTest extends
AbstractBasicIntegrationTest {
+ /** Cache used by KeyValue API tests. */
+ private static final String KEY_VALUE_CACHE_NAME = "bulk-operation-cache";
+
+ /** Number of entries processed by each bulk operation. */
+ private static final int ENTRY_COUNT = 30;
+
+ /** Number of person IDs in each tenant. */
+ private static final int PERSONS_PER_TENANT = 3;
+
+ /** Number of concurrent transaction workers. */
+ private static final int CONCURRENT_TX_THREADS = 6;
+
+ /** Duration of the concurrent load. */
+ private static final long CONCURRENT_TX_DURATION_MS =
TimeUnit.MINUTES.toMillis(1);
Review Comment:
This integration test adds a fixed 1-minute concurrent workload to the
Calcite integration suite, which can noticeably slow CI and increase flakiness.
Consider reducing the default duration (or making it configurable) while
keeping enough iterations to catch deadlocks.
##########
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();
+ Map<Object, CacheEntry<Object, Object>> entries =
+ lockBatches.computeIfAbsent(cache, key -> new
LinkedHashMap<>());
+ int keyColumnIdx = target.keyColumnIndex();
+
+ for (List<?> row : rows) {
+ Object key = row.get(keyColumnIdx);
+
+ // An outer join has no row to lock on its non-matching side.
+ if (key == null)
+ continue;
+
+ Object val = row.get(keyColumnIdx + 1);
+ GridCacheVersion ver = (GridCacheVersion)row.get(keyColumnIdx
+ 2);
+
+ // JOINs can repeat a row, but a transaction needs only one
lock per cache key.
+ entries.put(key, new CacheEntryImplEx<>(key, val, ver));
+ }
+ }
+
+ for (Map.Entry<IgniteInternalCache<Object, Object>, Map<Object,
CacheEntry<Object, Object>>> batch :
+ lockBatches.entrySet()) {
+ Map<Object, CacheEntry<Object, Object>> entries = batch.getValue();
+
+ orderLockEntries(batch.getKey().context(), entries);
+ }
+
+ return new ArrayList<>(lockBatches.entrySet());
+ }
+
+ /**
+ * Orders entries by partition, key hash, and serialized key bytes in case
of a hash collision.
+ *
+ * @param cctx Cache context used to prepare cache keys.
+ * @param entries Entries to order.
+ */
+ private static void orderLockEntries(
+ GridCacheContext<Object, Object> cctx,
+ Map<Object, CacheEntry<Object, Object>> entries
+ ) {
+ List<LockEntry> orderedEntries = new ArrayList<>(entries.size());
+
+ for (CacheEntry<Object, Object> entry : entries.values()) {
+ KeyCacheObject key = cctx.toCacheKeyObject(entry.getKey());
+
+ orderedEntries.add(new LockEntry(entry, key));
+ }
+
+ orderedEntries.sort(Comparator.comparingInt((LockEntry entry) ->
entry.part)
+ .thenComparingInt(entry -> entry.keyHash));
+
+ for (int start = 0; start < orderedEntries.size(); ) {
+ LockEntry first = orderedEntries.get(start);
+ int end = start + 1;
+
+ while (end < orderedEntries.size()
+ && orderedEntries.get(end).part == first.part
+ && orderedEntries.get(end).keyHash == first.keyHash)
+ end++;
+
+ if (end - start > 1) {
+ try {
+ for (int i = start; i < end; i++)
+
orderedEntries.get(i).prepareKeyBytes(cctx.cacheObjectContext());
+ }
+ catch (IgniteCheckedException e) {
+ throw new IgniteSQLException("Failed to serialize a cache
key for lock ordering", e);
+ }
+
+ orderedEntries.subList(start, end).sort(Comparator.comparing(
+ entry -> entry.keyBytes,
+ UnsignedBytes.lexicographicalComparator()
+ ));
+ }
+
+ start = end;
+ }
+
+ entries.clear();
+
+ for (LockEntry entry : orderedEntries)
+ entries.put(entry.entry.getKey(), entry.entry);
+ }
+
+ /** Cache entry with the key attributes used for lock ordering. */
+ private static class LockEntry {
+ /** Cache entry. */
+ private final CacheEntry<Object, Object> entry;
+
+ /** Key partition. */
+ private final int part;
+
+ /** Key hash. */
+ private final int keyHash;
+
+ /** Cache key. */
+ private final KeyCacheObject key;
+
+ /** Serialized key bytes, initialized only for hash collisions. */
+ private byte[] keyBytes;
+
+ /**
+ * @param entry Cache entry.
+ * @param key Cache key.
+ */
+ private LockEntry(CacheEntry<Object, Object> entry, KeyCacheObject
key) {
+ this.entry = entry;
+ this.key = key;
+ part = key.partition();
+ keyHash = key.hashCode();
+ }
+
+ /**
+ * Serializes the key for collision resolution.
+ *
+ * @param ctx Cache object context.
+ * @throws IgniteCheckedException If serialization fails.
+ */
+ private void prepareKeyBytes(CacheObjectValueContext ctx) throws
IgniteCheckedException {
+ keyBytes = key.valueBytes(ctx);
+ }
+ }
+
+ /**
+ * Tries to lock all collected entries within a transaction savepoint.
+ *
+ * @param userTx Transaction that acquires the locks.
+ * @param lockBatches Cache entries grouped by cache.
+ * @param waitMs Lock wait time in the internal representation.
+ * @param lockAcquisitionEndTime Absolute lock deadline, or {@code 0} to
use the transaction/query timeout.
+ * @return {@code true} if every lock was acquired.
+ */
+ private static boolean tryAcquireLocks(
+ GridNearTxLocal userTx,
+ List<Map.Entry<IgniteInternalCache<Object, Object>, Map<Object,
CacheEntry<Object, Object>>>> lockBatches,
+ long waitMs,
+ long lockAcquisitionEndTime
+ ) {
+ try {
+ // lockTxEntries() requires the transaction to be bound to the
current thread
+ // (it checks cctx.tm().threadLocalTx()). Resume it here and
suspend afterwards,
+ // following the same pattern as
ModifyNode.invokeInsideTransaction().
+ userTx.resume();
+
+ try {
+ // Create a savepoint so that a failed lock attempt can be
rolled back without aborting the whole tx.
+ String spName = "_for_update_" + UUID.randomUUID();
+
+ userTx.savepoint(spName, false);
+
+ boolean locked = true;
+
+ try {
+ for (Map.Entry<IgniteInternalCache<Object, Object>,
Map<Object, CacheEntry<Object, Object>>> batch :
+ lockBatches) {
+ if (batch.getValue().isEmpty())
+ continue;
+
+ long batchWaitMs = waitMs;
+
+ if (lockAcquisitionEndTime > 0) {
+ batchWaitMs = lockAcquisitionEndTime -
U.currentTimeMillis();
+
+ // Excluse case where
+ if (batchWaitMs == 0)
+ batchWaitMs = -1L;
+ }
+
+ if
(!batch.getKey().lockTxEntries(batch.getValue().values(), batchWaitMs)) {
+ locked = false;
+ break;
+ }
+ }
+ }
+ catch (IgniteCheckedException e) {
+ try {
+ userTx.rollbackToSavepoint(spName);
+ }
+ catch (Exception rollbackEx) {
+ e.addSuppressed(rollbackEx);
+ }
+
+ throw new IgniteSQLException("Failed to acquire locks for
SELECT FOR UPDATE",
+ IgniteQueryErrorCode.CONCURRENT_UPDATE, e);
+ }
+
+ if (!locked) {
+ try {
+ userTx.rollbackToSavepoint(spName);
+ }
+ catch (IgniteCheckedException rollbackEx) {
+ throw new IgniteSQLException("Failed to rollback
savepoint after lock failure",
+ IgniteQueryErrorCode.UNKNOWN, rollbackEx);
+ }
+
+ return false;
+ }
+
+ try {
+ userTx.releaseSavepoint(spName);
+ }
+ catch (IgniteCheckedException e) {
+ throw new IgniteSQLException("Failed to release savepoint
after successful lock",
+ IgniteQueryErrorCode.UNKNOWN, e);
+ }
+ }
+ finally {
+ userTx.suspend();
+ }
+ }
+ catch (IgniteCheckedException e) {
+ throw new IgniteSQLException("Failed to get cache entries for
SELECT FOR UPDATE",
+ IgniteQueryErrorCode.UNKNOWN, e);
+ }
+
+ return true;
+ }
+
+ /**
+ * Creates a cursor containing only user-visible columns and their
metadata.
+ *
+ * @param qry Root query providing the type factory.
+ * @param plan SELECT FOR UPDATE plan providing field metadata.
+ * @param rows Selected rows containing user-visible and internal lock
columns.
+ * @param userColCnt Number of user-visible columns.
+ * @return Cursor containing only user-visible data and metadata.
+ */
+ private FieldsQueryCursor<List<?>> createResultCursor(
+ RootQuery<Row> qry,
+ SelectForUpdatePlan plan,
+ List<List<?>> rows,
+ int userColCnt
+ ) {
+ List<List<?>> userRows = rows.isEmpty() ? Collections.emptyList() :
new ArrayList<>(rows.size());
+
+ for (List<?> row : rows)
+ userRows.add(row.subList(0, userColCnt));
+
Review Comment:
`row.subList(0, userColCnt)` returns a view backed by the full inner row
list, so the cursor will still retain references to hidden lock columns
(including `_VAL`, which can be large) and defeat the intent to return only
user-visible columns. Copy the user columns into a new list to allow hidden
columns to be GC’d.
--
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]
