vldpyatkov commented on code in PR #13366:
URL: https://github.com/apache/ignite/pull/13366#discussion_r3713130976
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExecutionServiceImpl.java:
##########
@@ -581,6 +598,228 @@ 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);
+
+ // waitSeconds: null = use tx remaining time (0), 0 = NOWAIT (-1),
positive = ms.
+ Long waitSeconds = plan.waitSeconds();
+ long waitMs;
+
+ if (waitSeconds == null)
+ waitMs = 0L;
+ else if (waitSeconds == 0L)
+ waitMs = -1L;
+ else
+ waitMs = waitSeconds * 1000L;
+
+ // 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();
+ qryReg.register(selectQry);
+ }
+ }
+
+ /**
+ * Executes the inner SELECT once and tries to lock the selected row
versions.
+ *
+ * @return Result cursor when locking succeeds, or {@code null} when the
SELECT must be executed again.
+ */
+ @Nullable private FieldsQueryCursor<List<?>> tryExecuteForUpdate(
+ RootQuery<Row> qry,
+ SelectForUpdatePlan plan,
+ GridNearTxLocal userTx,
+ long waitMs,
+ long deadline
+ ) {
+ // Run the inner SELECT (with _KEY, _VAL, _VER appended) and collect
all rows.
+ ListFieldsQueryCursor<?> innerCursor = mapAndExecutePlan(qry,
plan.innerPlan());
+ List<List<?>> rows = innerCursor.getAll();
+
+ int userColCnt = plan.userColumnCount();
+
+ if (rows.isEmpty()) {
+ // Nothing to lock – return an empty cursor with user-only field
metadata.
+ QueryCursorImpl<List<?>> resCur = new
QueryCursorImpl<>(Collections.emptyList(), null, false);
+
+ IgniteTypeFactory typeFactory = qry.context().typeFactory();
+ List<GridQueryFieldMetadata> meta =
plan.innerPlan().fieldsMetadata().queryFieldsMetadata(typeFactory);
+
+ resCur.fieldsMeta(meta.subList(0, userColCnt));
+
+ return resCur;
+ }
+
+ Map<IgniteInternalCache<Object, Object>, Map<Object,
CacheEntry<Object, Object>>> entriesByCache =
+ new LinkedHashMap<>();
+
+ 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 =
+ entriesByCache.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));
+ }
+ }
+
+ List<Map.Entry<IgniteInternalCache<Object, Object>, Map<Object,
CacheEntry<Object, Object>>>> lockBatches =
+ new ArrayList<>(entriesByCache.entrySet());
+
+ lockBatches.sort(Comparator.comparingInt(left ->
left.getKey().context().cacheId()));
+
+ 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 (deadline > 0) {
+ batchWaitMs = deadline - U.currentTimeMillis();
+
+ if (batchWaitMs <= 0)
+ batchWaitMs = -1L;
Review Comment:
Because we could lose time in the operation before. Unlikely we can get a
strict 0 here.
--
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]
