You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by vo...@apache.org on 2019/04/02 14:44:57 UTC
[ignite] branch master updated: IGNITE-10104: MVCC: Fixed SFU for
REPLICATED caches. This closes #6141.
This is an automated email from the ASF dual-hosted git repository.
vozerov pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 27ea976 IGNITE-10104: MVCC: Fixed SFU for REPLICATED caches. This closes #6141.
27ea976 is described below
commit 27ea976af2a485fee1c1f0713e224404d534614f
Author: rkondakov <ko...@mail.ru>
AuthorDate: Tue Apr 2 17:44:46 2019 +0300
IGNITE-10104: MVCC: Fixed SFU for REPLICATED caches. This closes #6141.
---
.../cache/distributed/dht/CompoundLockFuture.java | 112 ---
.../dht/GridDhtTxAbstractEnlistFuture.java | 8 +-
.../dht/GridDhtTxQueryEnlistFuture.java | 2 +-
.../near/GridNearTxAbstractEnlistFuture.java | 5 +-
.../near/GridNearTxSelectForUpdateFuture.java | 462 ------------
.../distributed/near/TxTopologyVersionFuture.java | 176 -----
.../cache/query/GridCacheTwoStepQuery.java | 26 +-
.../query/h2/DhtResultSetEnlistFuture.java | 63 --
.../processors/query/h2/H2FieldsIterator.java | 5 +-
.../processors/query/h2/H2KeyValueIterator.java | 2 +-
.../processors/query/h2/H2ResultSetIterator.java | 7 +-
.../internal/processors/query/h2/H2Utils.java | 15 -
.../processors/query/h2/IgniteH2Indexing.java | 472 ++++++------
.../query/h2/NearResultSetEnlistFuture.java | 59 --
.../internal/processors/query/h2/QueryParser.java | 82 ++-
.../query/h2/QueryParserResultSelect.java | 61 +-
.../processors/query/h2/ResultSetEnlistFuture.java | 136 ----
.../processors/query/h2/dml/UpdatePlanBuilder.java | 12 +-
.../query/h2/sql/GridSqlQueryParser.java | 55 +-
.../query/h2/sql/GridSqlQuerySplitter.java | 84 ++-
.../processors/query/h2/sql/GridSqlSelect.java | 24 +
.../query/h2/twostep/GridMapQueryExecutor.java | 197 +----
.../query/h2/twostep/GridReduceQueryExecutor.java | 157 +---
.../query/h2/twostep/MapQueryResults.java | 15 +-
.../query/h2/twostep/ReduceQueryRun.java | 14 -
.../query/h2/twostep/ReduceResultPage.java | 20 +-
...cheMvccPartitionedSelectForUpdateQueryTest.java | 47 --
...acheMvccReplicatedSelectForUpdateQueryTest.java | 30 -
.../CacheMvccSelectForUpdateQueryAbstractTest.java | 370 ----------
.../CacheMvccSelectForUpdateQueryBasicTest.java | 789 +++++++++++++++++++++
.../mvcc/CacheMvccSelectForUpdateQueryTest.java | 184 +++++
.../IgniteBinaryCacheQueryTestSuite2.java | 3 +
.../testsuites/IgniteCacheMvccSqlTestSuite.java | 9 +-
33 files changed, 1468 insertions(+), 2235 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/CompoundLockFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/CompoundLockFuture.java
deleted file mode 100644
index 63eff40..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/CompoundLockFuture.java
+++ /dev/null
@@ -1,112 +0,0 @@
-/*
- * 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.cache.distributed.dht;
-
-import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
-import org.apache.ignite.internal.IgniteInternalFuture;
-import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxAbstractEnlistFuture;
-import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxSelectForUpdateFuture;
-import org.apache.ignite.internal.util.future.GridFutureAdapter;
-import org.apache.ignite.lang.IgniteInClosure;
-
-/**
- *
- */
-public class CompoundLockFuture extends GridFutureAdapter<Void> implements DhtLockFuture<Void>, IgniteInClosure<IgniteInternalFuture<?>> {
- /** */
- private static final long serialVersionUID = 4644646033267042131L;
- /** */
- private static final AtomicIntegerFieldUpdater<CompoundLockFuture> CNT_UPD =
- AtomicIntegerFieldUpdater.newUpdater(CompoundLockFuture.class, "cnt");
- /** */
- private volatile int cnt;
- /** */
- private final GridDhtTxLocalAdapter tx;
-
- /**
- * @param cnt ResultSet futures count.
- * @param tx Transaction.
- */
- public CompoundLockFuture(int cnt, GridDhtTxLocalAdapter tx) {
- this.cnt = cnt;
- this.tx = tx;
- }
-
- /**
- * @param fut ResultSet future.
- */
- public void register(IgniteInternalFuture<?> fut) {
- fut.listen(this);
- }
-
- /**
- * Init method.
- */
- public void init() {
- while(true) {
- IgniteInternalFuture<?> fut = tx.lockFuture();
-
- if (fut == GridDhtTxLocalAdapter.ROLLBACK_FUT) {
- onDone(tx.timedOut() ? tx.timeoutException() : tx.rollbackException());
-
- break;
- }
- else if (fut != null) {
- // Wait for previous future.
- assert fut instanceof GridNearTxAbstractEnlistFuture
- || fut instanceof GridDhtTxAbstractEnlistFuture
- || fut instanceof CompoundLockFuture
- || fut instanceof GridNearTxSelectForUpdateFuture : fut;
-
- // Terminate this future if parent future is terminated by rollback.
- if (!fut.isDone()) {
- fut.listen(new IgniteInClosure<IgniteInternalFuture>() {
- @Override public void apply(IgniteInternalFuture fut) {
- if (fut.error() != null)
- onDone(fut.error());
- }
- });
- }
- else if (fut.error() != null)
- onDone(fut.error());
-
- break;
- }
- else if (tx.updateLockFuture(null, this))
- break;
- }
- }
-
- @Override public void apply(IgniteInternalFuture<?> future) {
- if (!isDone() && (future.error() != null || CNT_UPD.decrementAndGet(this) == 0)) {
- Throwable err = future.error();
-
- if (err == null)
- tx.clearLockFuture(this);
-
- onDone(err);
- }
- }
-
- /** {@inheritDoc} */
- @Override public void onError(Throwable error) {
- assert error != null;
-
- onDone(error);
- }
-}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxAbstractEnlistFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxAbstractEnlistFuture.java
index 6b443cd..d1bd295 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxAbstractEnlistFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxAbstractEnlistFuture.java
@@ -49,7 +49,6 @@ import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTxMapping;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxAbstractEnlistFuture;
-import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxSelectForUpdateFuture;
import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinator;
import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
import org.apache.ignite.internal.processors.cache.mvcc.MvccUtils;
@@ -294,9 +293,7 @@ public abstract class GridDhtTxAbstractEnlistFuture<T> extends GridCacheFutureAd
else if (fut != null) {
// Wait for previous future.
assert fut instanceof GridNearTxAbstractEnlistFuture
- || fut instanceof GridDhtTxAbstractEnlistFuture
- || fut instanceof CompoundLockFuture
- || fut instanceof GridNearTxSelectForUpdateFuture : fut;
+ || fut instanceof GridDhtTxAbstractEnlistFuture : fut;
// Terminate this future if parent future is terminated by rollback.
if (!fut.isDone()) {
@@ -641,7 +638,8 @@ public abstract class GridDhtTxAbstractEnlistFuture<T> extends GridCacheFutureAd
assert updRes != null && updRes.updateFuture() == null;
- onEntryProcessed(entry.key(), updRes);
+ if (op != EnlistOperation.LOCK)
+ onEntryProcessed(entry.key(), updRes);
if (!updRes.success()
|| updRes.filtered()
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxQueryEnlistFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxQueryEnlistFuture.java
index ee49c6b..4d56344 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxQueryEnlistFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxQueryEnlistFuture.java
@@ -130,7 +130,7 @@ public final class GridDhtTxQueryEnlistFuture extends GridDhtTxQueryAbstractEnli
params,
flags,
pageSize,
- 0,
+ (int)timeout,
tx.topologyVersionSnapshot(),
mvccSnapshot,
new GridQueryCancel()
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxAbstractEnlistFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxAbstractEnlistFuture.java
index 89e7761..379c13b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxAbstractEnlistFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxAbstractEnlistFuture.java
@@ -34,7 +34,6 @@ import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
import org.apache.ignite.internal.processors.cache.GridCacheMvccCandidate;
import org.apache.ignite.internal.processors.cache.GridCacheVersionedFuture;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTxMapping;
-import org.apache.ignite.internal.processors.cache.distributed.dht.CompoundLockFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTopologyFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxAbstractEnlistFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxLocalAdapter;
@@ -155,9 +154,7 @@ public abstract class GridNearTxAbstractEnlistFuture<T> extends GridCacheCompoun
else if (fut != null) {
// Wait for previous future.
assert fut instanceof GridNearTxAbstractEnlistFuture
- || fut instanceof GridDhtTxAbstractEnlistFuture
- || fut instanceof CompoundLockFuture
- || fut instanceof GridNearTxSelectForUpdateFuture : fut;
+ || fut instanceof GridDhtTxAbstractEnlistFuture : fut;
// Terminate this future if parent future is terminated by rollback.
if (!fut.isDone()) {
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxSelectForUpdateFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxSelectForUpdateFuture.java
deleted file mode 100644
index e513a44..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxSelectForUpdateFuture.java
+++ /dev/null
@@ -1,462 +0,0 @@
-/*
- * 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.cache.distributed.near;
-
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.UUID;
-import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
-import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
-import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.IgniteLogger;
-import org.apache.ignite.cluster.ClusterNode;
-import org.apache.ignite.internal.IgniteInternalFuture;
-import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
-import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
-import org.apache.ignite.internal.processors.cache.GridCacheCompoundIdentityFuture;
-import org.apache.ignite.internal.processors.cache.GridCacheContext;
-import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
-import org.apache.ignite.internal.processors.cache.GridCacheMvccCandidate;
-import org.apache.ignite.internal.processors.cache.GridCacheVersionedFuture;
-import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTxMapping;
-import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
-import org.apache.ignite.internal.processors.timeout.GridTimeoutObjectAdapter;
-import org.apache.ignite.internal.transactions.IgniteTxTimeoutCheckedException;
-import org.apache.ignite.internal.util.future.GridFutureAdapter;
-import org.apache.ignite.internal.util.tostring.GridToStringExclude;
-import org.apache.ignite.internal.util.typedef.X;
-import org.apache.ignite.internal.util.typedef.internal.CU;
-import org.apache.ignite.internal.util.typedef.internal.S;
-import org.apache.ignite.lang.IgniteUuid;
-import org.jetbrains.annotations.NotNull;
-import org.jetbrains.annotations.Nullable;
-
-/**
- * A future tracking requests for remote nodes transaction enlisting and locking
- * of entries produced with complex DML queries requiring reduce step.
- */
-public class GridNearTxSelectForUpdateFuture extends GridCacheCompoundIdentityFuture<Long>
- implements GridCacheVersionedFuture<Long> {
- /** */
- private static final long serialVersionUID = 6931664882548658420L;
-
- /** Done field updater. */
- private static final AtomicIntegerFieldUpdater<GridNearTxSelectForUpdateFuture> DONE_UPD =
- AtomicIntegerFieldUpdater.newUpdater(GridNearTxSelectForUpdateFuture.class, "done");
-
- /** Done field updater. */
- private static final AtomicReferenceFieldUpdater<GridNearTxSelectForUpdateFuture, Throwable> EX_UPD =
- AtomicReferenceFieldUpdater.newUpdater(GridNearTxSelectForUpdateFuture.class, Throwable.class, "ex");
-
- /** */
- @SuppressWarnings("unused")
- @GridToStringExclude
- private volatile int done;
-
- /** */
- @SuppressWarnings("unused")
- @GridToStringExclude
- private volatile Throwable ex;
-
- /** Cache context. */
- @GridToStringExclude
- private final GridCacheContext<?, ?> cctx;
-
- /** Transaction. */
- private final GridNearTxLocal tx;
-
- /** Mvcc future id. */
- private final IgniteUuid futId;
-
- /** Lock version. */
- private final GridCacheVersion lockVer;
-
- /** */
- private AffinityTopologyVersion topVer;
-
- /** */
- private final long timeout;
-
- /** Logger. */
- @GridToStringExclude
- private final IgniteLogger log;
-
- /** Timeout object. */
- @GridToStringExclude
- private LockTimeoutObject timeoutObj;
-
- /** Ids of mini futures. */
- private final Map<UUID, Integer> miniFutIds = new HashMap<>();
-
- /**
- * @param cctx Cache context.
- * @param tx Transaction.
- * @param timeout Timeout.
- */
- public GridNearTxSelectForUpdateFuture(GridCacheContext<?, ?> cctx,
- GridNearTxLocal tx,
- long timeout) {
- super(CU.longReducer());
-
- this.cctx = cctx;
- this.tx = tx;
- this.timeout = timeout;
-
- futId = IgniteUuid.randomUuid();
- lockVer = tx.xidVersion();
-
- log = cctx.logger(GridNearTxSelectForUpdateFuture.class);
- }
-
- /**
- * @return Cache context.
- */
- public GridCacheContext<?, ?> cache() {
- return cctx;
- }
-
- /**
- * @param node Node.
- */
- private void map(ClusterNode node) {
- GridDistributedTxMapping mapping = tx.mappings().get(node.id());
-
- if (mapping == null)
- tx.mappings().put(mapping = new GridDistributedTxMapping(node));
-
- mapping.markQueryUpdate();
-
- if (node.isLocal())
- tx.colocatedLocallyMapped(true);
-
- int futId = futuresCountNoLock();
-
- miniFutIds.put(node.id(), futId);
-
- add(new NodeFuture(node));
- }
-
- /**
- * Process result of query execution on given
- * @param nodeId Node id.
- * @param cnt Total rows counter on given node.
- * @param removeMapping Whether transaction mapping should be removed for node.
- * @param err Error.
- */
- public void onResult(UUID nodeId, Long cnt, boolean removeMapping, @Nullable Throwable err) {
- NodeFuture nodeFut = mapFuture(nodeId);
-
- if (nodeFut != null)
- nodeFut.onResult(cnt, removeMapping, err);
- }
-
- /** {@inheritDoc} */
- @Override protected boolean processFailure(Throwable err, IgniteInternalFuture<Long> fut) {
- if (ex != null || !EX_UPD.compareAndSet(this, null, err))
- ex.addSuppressed(err);
-
- return true;
- }
-
- /** {@inheritDoc} */
- @Override public boolean onDone(@Nullable Long res, @Nullable Throwable err, boolean cancelled) {
- if (!DONE_UPD.compareAndSet(this, 0, 1))
- return false;
-
- cctx.tm().resetContext();
-
- Throwable ex0 = ex;
-
- if (ex0 != null) {
- if (err != null)
- ex0.addSuppressed(err);
-
- err = ex0;
- }
-
- if (!cancelled && err == null)
- tx.clearLockFuture(this);
- else
- tx.setRollbackOnly();
-
- boolean done = super.onDone(res, err, cancelled);
-
- assert done;
-
- // Clean up.
- cctx.mvcc().removeVersionedFuture(this);
-
- if (timeoutObj != null)
- cctx.time().removeTimeoutObject(timeoutObj);
-
- return true;
- }
-
- /**
- * Finds pending map node future by the given ID.
- *
- * @param nodeId Node id.
- * @return Map node future.
- */
- private NodeFuture mapFuture(UUID nodeId) {
- synchronized (this) {
- Integer idx = miniFutIds.get(nodeId);
-
- if (idx == null)
- throw new IllegalStateException("SELECT FOR UPDATE node future not found [nodeId=" + nodeId + "].");
-
- assert idx >= 0 && idx < futuresCountNoLock();
-
- IgniteInternalFuture<Long> fut = future(idx);
-
- if (!fut.isDone())
- return (NodeFuture)fut;
- }
-
- return null;
- }
-
- /** {@inheritDoc} */
- @Override public GridCacheVersion version() {
- return lockVer;
- }
-
- /** {@inheritDoc} */
- @Override public boolean onOwnerChanged(GridCacheEntryEx entry, GridCacheMvccCandidate owner) {
- return false;
- }
-
- /** {@inheritDoc} */
- @Override public IgniteUuid futureId() {
- return futId;
- }
-
- /** {@inheritDoc} */
- @Override public boolean onNodeLeft(UUID nodeId) {
- if (topVer == null)
- return false; // Local query, do nothing.
-
- for (IgniteInternalFuture<?> fut : futures()) {
- NodeFuture f = (NodeFuture)fut;
-
- if (f.node.id().equals(nodeId)) {
- if (log.isDebugEnabled())
- log.debug("Found mini-future for left node [nodeId=" + nodeId + ", mini=" + f + ", fut=" +
- this + ']');
-
- ClusterTopologyCheckedException topEx = new ClusterTopologyCheckedException("Failed to enlist keys " +
- "(primary node left grid, retry transaction if possible) [node=" + nodeId + ']');
-
- topEx.retryReadyFuture(cctx.shared().nextAffinityReadyFuture(topVer));
-
- return f.onResult(0, false, topEx);
- }
- }
-
- if (log.isDebugEnabled())
- log.debug("Future does not have mapping for left node (ignoring) [nodeId=" + nodeId +
- ", fut=" + this + ']');
-
- return false;
- }
-
- /** {@inheritDoc} */
- @Override public boolean trackable() {
- return true;
- }
-
- /** {@inheritDoc} */
- @Override public void markNotTrackable() {
- // No-op.
- }
-
- /** {@inheritDoc} */
- @Override protected void logError(IgniteLogger log, String msg, Throwable e) {
- // no-op
- }
-
- /** {@inheritDoc} */
- @Override protected void logDebug(IgniteLogger log, String msg) {
- // no-op
- }
-
- /** {@inheritDoc} */
- @Override public String toString() {
- return S.toString(GridNearTxSelectForUpdateFuture.class, this, super.toString());
- }
-
- /**
- * Initialize this future for distributed execution.
- * @param topVer Topology version.
- * @param nodes Nodes to run query on.
- */
- public synchronized void init(AffinityTopologyVersion topVer, Collection<ClusterNode> nodes) {
- doInit(topVer, nodes, false);
- }
-
- /**
- * Initialize this future for local execution.
- */
- public synchronized void initLocal() {
- doInit(null, Collections.singletonList(cctx.localNode()), true);
- }
-
- /**
- * Initialize this future for distributed or local execution.
- * @param topVer Topology version ({@code null} for local case).
- * @param nodes Nodes to run query on.
- * @param loc Local query flag.
- */
- private void doInit(@Nullable AffinityTopologyVersion topVer, Collection<ClusterNode> nodes, boolean loc) {
- assert !loc || (topVer == null && nodes.size() == 1 && nodes.iterator().next().isLocal());
- if (initialized())
- throw new IllegalStateException("SELECT FOR UPDATE future has been initialized already.");
-
- tx.init();
-
- if (timeout < 0) {
- // Time is out.
- onDone(timeoutException());
-
- return;
- }
- else if (timeout > 0)
- timeoutObj = new LockTimeoutObject();
-
- if (!tx.updateLockFuture(null, this)) {
- onDone(tx.timedOut() ? tx.timeoutException() : tx.rollbackException());
-
- return;
- }
-
- boolean added = cctx.mvcc().addFuture(this);
-
- assert added : this;
-
- try {
- tx.addActiveCache(cctx, false);
- }
- catch (IgniteCheckedException e) {
- onDone(e);
-
- return;
- }
-
- if (timeoutObj != null)
- cctx.time().addTimeoutObject(timeoutObj);
-
- this.topVer = topVer;
-
- for (ClusterNode n : nodes)
- map(n);
-
- markInitialized();
- }
-
- /**
- * @return Timeout exception.
- */
- @NotNull private IgniteTxTimeoutCheckedException timeoutException() {
- return new IgniteTxTimeoutCheckedException("Failed to acquire lock within provided timeout for " +
- "transaction [timeout=" + timeout + ", tx=" + tx + ']');
- }
-
- /**
- * A future tracking a single MAP request to be enlisted in transaction and locked on data node.
- */
- private class NodeFuture extends GridFutureAdapter<Long> {
- /** */
- private boolean completed;
-
- /** Node ID. */
- @GridToStringExclude
- private final ClusterNode node;
-
- /**
- * @param node Cluster node.
- *
- */
- private NodeFuture(ClusterNode node) {
- this.node = node;
- }
-
- /**
- * @return Node.
- */
- public ClusterNode node() {
- return node;
- }
-
- /**
- * @param cnt Total rows counter on given node.
- * @param removeMapping Whether transaction mapping should be removed for node.
- * @param err Exception.
- * @return {@code True} if future was completed by this call.
- */
- public boolean onResult(long cnt, boolean removeMapping, Throwable err) {
- synchronized (this) {
- if (completed)
- return false;
-
- completed = true;
- }
-
- if (X.hasCause(err, ClusterTopologyCheckedException.class) || removeMapping) {
- GridDistributedTxMapping m = tx.mappings().get(node.id());
-
- assert m != null && m.empty();
-
- tx.removeMapping(node.id());
-
- if (node.isLocal())
- tx.colocatedLocallyMapped(false);
- }
- else if (err == null && cnt > 0 && !node.isLocal())
- tx.hasRemoteLocks(true);
-
- return onDone(cnt, err);
- }
- }
-
- /**
- * Lock request timeout object.
- */
- private class LockTimeoutObject extends GridTimeoutObjectAdapter {
- /**
- * Default constructor.
- */
- LockTimeoutObject() {
- super(timeout);
- }
-
- /** {@inheritDoc} */
- @Override public void onTimeout() {
- if (log.isDebugEnabled())
- log.debug("Timed out waiting for lock response: " + this);
-
- onDone(timeoutException());
- }
-
- /** {@inheritDoc} */
- @Override public String toString() {
- return S.toString(LockTimeoutObject.class, this);
- }
- }
-}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/TxTopologyVersionFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/TxTopologyVersionFuture.java
deleted file mode 100644
index 345e742..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/TxTopologyVersionFuture.java
+++ /dev/null
@@ -1,176 +0,0 @@
-/*
- * 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.cache.distributed.near;
-
-import org.apache.ignite.IgniteCacheRestartingException;
-import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.internal.IgniteInternalFuture;
-import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
-import org.apache.ignite.internal.processors.cache.CacheStoppedException;
-import org.apache.ignite.internal.processors.cache.GridCacheContext;
-import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTopologyFuture;
-import org.apache.ignite.internal.util.future.GridFutureAdapter;
-import org.apache.ignite.internal.util.typedef.CI1;
-import org.apache.ignite.internal.util.typedef.internal.S;
-
-/**
- * Future to obtain/lock topology version for SELECT FOR UPDATE.
- */
-public class TxTopologyVersionFuture extends GridFutureAdapter<AffinityTopologyVersion> {
- /** Transaction. */
- private final GridNearTxLocal tx;
-
- /** Target cache context. */
- private final GridCacheContext<?, ?> cctx;
-
- /** Topology locked flag. */
- private boolean topLocked;
-
- /**
- * @param tx Transaction.
- * @param cctx Target cache context.
- */
- public TxTopologyVersionFuture(GridNearTxLocal tx, GridCacheContext cctx) {
- this.tx = tx;
- this.cctx = cctx;
-
- init();
- }
-
- /** */
- private void init() {
- // Obtain the topology version to use.
- long threadId = Thread.currentThread().getId();
-
- AffinityTopologyVersion topVer = cctx.mvcc().lastExplicitLockTopologyVersion(threadId);
-
- // If there is another system transaction in progress, use it's topology version to prevent deadlock.
- if (topVer == null && tx.system())
- topVer = cctx.tm().lockedTopologyVersion(threadId, tx);
-
- if (topVer != null)
- tx.topologyVersion(topVer);
-
- if (topVer == null)
- topVer = tx.topologyVersionSnapshot();
-
- if (topVer != null) {
- for (GridDhtTopologyFuture fut : cctx.shared().exchange().exchangeFutures()) {
- if (fut.exchangeDone() && fut.topologyVersion().equals(topVer)) {
- Throwable err = null;
-
- // Before cache validation, make sure that this topology future is already completed.
- try {
- fut.get();
- }
- catch (IgniteCheckedException e) {
- err = fut.error();
- }
-
- if (err == null)
- err = fut.validateCache(cctx, false, false, null, null);
-
- if (err != null) {
- onDone(err);
-
- return;
- }
-
- break;
- }
- }
-
- onDone(topVer);
-
- topLocked = true;
-
- return;
- }
-
- acquireTopologyVersion();
- }
-
- /**
- * Acquire topology future and wait for its completion.
- */
- private void acquireTopologyVersion() {
- cctx.topology().readLock();
-
- try {
- if (cctx.topology().stopping()) {
- onDone(
- cctx.shared().cache().isCacheRestarting(cctx.name())?
- new IgniteCacheRestartingException(cctx.name()):
- new CacheStoppedException(cctx.name()));
-
- return;
- }
-
- GridDhtTopologyFuture fut = cctx.topologyVersionFuture();
-
- if (fut.isDone()) {
- Throwable err = fut.validateCache(cctx, false, false, null, null);
-
- if (err != null) {
- onDone(err);
-
- return;
- }
-
- AffinityTopologyVersion topVer = fut.topologyVersion();
-
- if (tx != null)
- tx.topologyVersion(topVer);
-
- onDone(topVer);
- }
- else {
- fut.listen(new CI1<IgniteInternalFuture<AffinityTopologyVersion>>() {
- @Override public void apply(IgniteInternalFuture<AffinityTopologyVersion> fut) {
- try {
- fut.get();
-
- acquireTopologyVersion();
- }
- catch (IgniteCheckedException e) {
- onDone(e);
- }
- finally {
- cctx.shared().txContextReset();
- }
- }
- });
- }
- }
- finally {
- cctx.topology().readUnlock();
- }
- }
-
- /**
- * @return Client first flag.
- */
- public boolean clientFirst() {
- return cctx.localNode().isClient() && !topLocked && !tx.hasRemoteLocks();
- }
-
- /** {@inheritDoc} */
- @Override public String toString() {
- return S.toString(TxTopologyVersionFuture.class, this, super.toString());
- }
-}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheTwoStepQuery.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheTwoStepQuery.java
index 0fcddeb..c80484a 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheTwoStepQuery.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheTwoStepQuery.java
@@ -64,15 +64,23 @@ public class GridCacheTwoStepQuery {
/** */
private final boolean mvccEnabled;
- /** {@code FOR UPDATE} flag. */
- private final boolean forUpdate;
-
/** Number of positional arguments in the sql. */
private final int paramsCnt;
/**
- * @param originalSql Original query SQL.
- * @param tbls Tables in query.
+ *
+ * @param originalSql Original SQL.
+ * @param paramsCnt Parameters count.
+ * @param tbls Tables.
+ * @param rdc Reduce query.
+ * @param mapQrys Map query.
+ * @param skipMergeTbl Skip merge table flag.
+ * @param explain Explain flag.
+ * @param distributedJoins Distributed joins flag.
+ * @param derivedPartitions Derived partitions.
+ * @param cacheIds Cache ids.
+ * @param mvccEnabled Mvcc flag.
+ * @param locSplit Local split flag.
*/
public GridCacheTwoStepQuery(
String originalSql,
@@ -83,7 +91,6 @@ public class GridCacheTwoStepQuery {
boolean skipMergeTbl,
boolean explain,
boolean distributedJoins,
- boolean forUpdate,
PartitionResult derivedPartitions,
List<Integer> cacheIds,
boolean mvccEnabled,
@@ -97,7 +104,6 @@ public class GridCacheTwoStepQuery {
this.skipMergeTbl = skipMergeTbl;
this.explain = explain;
this.distributedJoins = distributedJoins;
- this.forUpdate = forUpdate;
this.derivedPartitions = derivedPartitions;
this.cacheIds = cacheIds;
this.mvccEnabled = mvccEnabled;
@@ -218,12 +224,6 @@ public class GridCacheTwoStepQuery {
return mvccEnabled;
}
- /**
- * @return {@code FOR UPDATE} flag.
- */
- public boolean forUpdate() {
- return forUpdate;
- }
/**
* @return Number of parameters
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/DhtResultSetEnlistFuture.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/DhtResultSetEnlistFuture.java
deleted file mode 100644
index 375789f..0000000
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/DhtResultSetEnlistFuture.java
+++ /dev/null
@@ -1,63 +0,0 @@
-/*
- * 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.h2;
-
-import java.sql.ResultSet;
-import java.util.UUID;
-import org.apache.ignite.internal.processors.cache.GridCacheContext;
-import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxLocalAdapter;
-import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxQueryAbstractEnlistFuture;
-import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
-import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
-import org.apache.ignite.internal.processors.query.UpdateSourceIterator;
-import org.apache.ignite.lang.IgniteUuid;
-import org.jetbrains.annotations.Nullable;
-
-/**
- *
- */
-public class DhtResultSetEnlistFuture extends GridDhtTxQueryAbstractEnlistFuture implements ResultSetEnlistFuture {
- /** */
- private ResultSet rs;
-
- /**
- * @param nearNodeId Near node ID.
- * @param nearLockVer Near lock version.
- * @param mvccSnapshot Mvcc snapshot.
- * @param threadId Thread ID.
- * @param nearFutId Near future id.
- * @param nearMiniId Near mini future id.
- * @param parts Partitions.
- * @param tx Transaction.
- * @param timeout Lock acquisition timeout.
- * @param cctx Cache context.
- * @param rs Result set to process.
- */
- public DhtResultSetEnlistFuture(UUID nearNodeId, GridCacheVersion nearLockVer,
- MvccSnapshot mvccSnapshot, long threadId, IgniteUuid nearFutId, int nearMiniId, @Nullable int[] parts,
- GridDhtTxLocalAdapter tx, long timeout, GridCacheContext<?, ?> cctx, ResultSet rs) {
- super(nearNodeId, nearLockVer, mvccSnapshot, threadId, nearFutId, nearMiniId, tx, timeout, cctx);
-
- this.rs = rs;
- }
-
- /** {@inheritDoc} */
- @Override protected UpdateSourceIterator<?> createIterator() {
- return ResultSetEnlistFuture.createIterator(rs);
- }
-}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2FieldsIterator.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2FieldsIterator.java
index cdafb65..7ac85a6 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2FieldsIterator.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2FieldsIterator.java
@@ -40,14 +40,13 @@ public class H2FieldsIterator extends H2ResultSetIterator<List<?>> {
/**
* @param data Data.
* @param mvccTracker Mvcc tracker.
- * @param forUpdate {@code SELECT FOR UPDATE} flag.
* @param detachedConn Detached connection.
* @throws IgniteCheckedException If failed.
*/
- public H2FieldsIterator(ResultSet data, MvccQueryTracker mvccTracker, boolean forUpdate,
+ public H2FieldsIterator(ResultSet data, MvccQueryTracker mvccTracker,
ThreadLocalObjectPool<H2ConnectionWrapper>.Reusable detachedConn)
throws IgniteCheckedException {
- super(data, forUpdate);
+ super(data);
assert detachedConn != null;
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2KeyValueIterator.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2KeyValueIterator.java
index 770d9d5..31add34 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2KeyValueIterator.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2KeyValueIterator.java
@@ -34,7 +34,7 @@ public class H2KeyValueIterator<K, V> extends H2ResultSetIterator<IgniteBiTuple<
* @throws IgniteCheckedException If failed.
*/
protected H2KeyValueIterator(ResultSet data) throws IgniteCheckedException {
- super(data, false);
+ super(data);
}
/** {@inheritDoc} */
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2ResultSetIterator.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2ResultSetIterator.java
index 6f75603..cfc1dd2 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2ResultSetIterator.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2ResultSetIterator.java
@@ -68,10 +68,9 @@ public abstract class H2ResultSetIterator<T> extends GridCloseableIteratorAdapte
/**
* @param data Data array.
- * @param forUpdate Whether is result is one of {@code SELECT FOR UPDATE} query.
* @throws IgniteCheckedException If failed.
*/
- protected H2ResultSetIterator(ResultSet data, boolean forUpdate) throws IgniteCheckedException {
+ protected H2ResultSetIterator(ResultSet data) throws IgniteCheckedException {
this.data = data;
try {
@@ -83,9 +82,7 @@ public abstract class H2ResultSetIterator<T> extends GridCloseableIteratorAdapte
if (data != null) {
try {
- int colsCnt = data.getMetaData().getColumnCount();
-
- row = new Object[forUpdate ? colsCnt - 1 : colsCnt];
+ row = new Object[data.getMetaData().getColumnCount()];
}
catch (SQLException e) {
throw new IgniteCheckedException(e);
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2Utils.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2Utils.java
index d4430ea..48bed10 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2Utils.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2Utils.java
@@ -833,16 +833,12 @@ public class H2Utils {
*
* @param idx Indexing.
* @param cacheIds Cache IDs.
- * @param mvccEnabled MVCC enabled flag.
- * @param forUpdate For update flag.
* @param tbls Tables.
*/
@SuppressWarnings("ForLoopReplaceableByForEach")
public static void checkQuery(
IgniteH2Indexing idx,
List<Integer> cacheIds,
- boolean mvccEnabled,
- boolean forUpdate,
Collection<QueryTable> tbls
) {
GridCacheSharedContext sharedCtx = idx.kernalContext().cache().context();
@@ -868,17 +864,6 @@ public class H2Utils {
}
}
- // Check FOR UPDATE invariants: only one table, MVCC is there.
- if (forUpdate) {
- if (cacheIds.size() != 1)
- throw new IgniteSQLException("SELECT FOR UPDATE is supported only for queries " +
- "that involve single transactional cache.");
-
- if (!mvccEnabled)
- throw new IgniteSQLException("SELECT FOR UPDATE query requires transactional cache " +
- "with MVCC enabled.", IgniteQueryErrorCode.UNSUPPORTED_OPERATION);
- }
-
// Check for joins between system views and normal tables.
if (!F.isEmpty(tbls)) {
for (QueryTable tbl : tbls) {
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java
index 7598bb0..2434e27 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java
@@ -59,11 +59,10 @@ import org.apache.ignite.internal.processors.cache.CacheOperationContext;
import org.apache.ignite.internal.processors.cache.DynamicCacheDescriptor;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheContextInfo;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.QueryCursorImpl;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
-import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxSelectForUpdateFuture;
-import org.apache.ignite.internal.processors.cache.distributed.near.TxTopologyVersionFuture;
import org.apache.ignite.internal.processors.cache.mvcc.MvccQueryTracker;
import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
import org.apache.ignite.internal.processors.cache.mvcc.MvccUtils;
@@ -144,7 +143,6 @@ import org.apache.ignite.lang.IgniteBiClosure;
import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.lang.IgniteFuture;
import org.apache.ignite.lang.IgniteInClosure;
-import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.marshaller.Marshaller;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.plugin.security.SecurityPermission;
@@ -160,6 +158,8 @@ import org.h2.util.JdbcUtils;
import org.jetbrains.annotations.Nullable;
import static java.lang.Boolean.FALSE;
+import static org.apache.ignite.IgniteSystemProperties.IGNITE_MVCC_TX_SIZE_CACHING_THRESHOLD;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccCachingManager.TX_SIZE_THRESHOLD;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.checkActive;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.mvccEnabled;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.requestSnapshot;
@@ -429,9 +429,10 @@ public class IgniteH2Indexing implements GridQueryIndexing {
* @param qryParams Query parameters.
* @param select Select.
* @param filter Cache name and key filter.
- * @param autoStartTx Start transaction flag.
* @param mvccTracker Query tracker.
* @param cancel Query cancel.
+ * @param inTx Flag whether the query is executed in transaction.
+ * @param timeout Timeout.
* @return Query result.
* @throws IgniteCheckedException If failed.
*/
@@ -440,13 +441,19 @@ public class IgniteH2Indexing implements GridQueryIndexing {
QueryParameters qryParams,
QueryParserResultSelect select,
final IndexingQueryFilter filter,
- boolean autoStartTx,
MvccQueryTracker mvccTracker,
- GridQueryCancel cancel
+ GridQueryCancel cancel,
+ boolean inTx,
+ int timeout
) throws IgniteCheckedException {
- String qry = qryDesc.sql();
+ assert !select.mvccEnabled() || mvccTracker != null;
+
+ String qry;
- GridNearTxLocal tx = null;
+ if (select.forUpdate())
+ qry = inTx ? select.forUpdateQueryTx() : select.forUpdateQueryOutTx();
+ else
+ qry = qryDesc.sql();
boolean mvccEnabled = mvccEnabled(kernalContext());
@@ -455,71 +462,13 @@ public class IgniteH2Indexing implements GridQueryIndexing {
try {
assert select != null;
- MvccSnapshot mvccSnapshot = null;
-
- boolean forUpdate = select.forUpdate();
-
- if (forUpdate && !mvccEnabled)
- throw new IgniteSQLException("SELECT FOR UPDATE query requires transactional " +
- "cache with MVCC enabled.", IgniteQueryErrorCode.UNSUPPORTED_OPERATION);
-
- GridNearTxSelectForUpdateFuture sfuFut = null;
-
- int opTimeout = qryParams.timeout();
-
- if (mvccEnabled) {
- if (mvccTracker == null) {
- Integer mvccCacheId = select.mvccCacheId();
-
- if (mvccCacheId != null) {
- GridCacheContext mvccCacheCtx = ctx.cache().context().cacheContext(select.mvccCacheId());
-
- if (mvccCacheCtx == null)
- throw new IgniteCheckedException("Cache has been stopped concurrently [cacheId=" +
- mvccCacheId + ']');
-
- mvccTracker = MvccUtils.mvccTracker(mvccCacheCtx, autoStartTx);
- }
- }
-
- if (mvccTracker != null) {
- mvccSnapshot = mvccTracker.snapshot();
-
- tx = checkActive(tx(ctx));
-
- opTimeout = operationTimeout(opTimeout, tx);
- }
-
- if (forUpdate) {
- if (mvccTracker == null)
- throw new IgniteSQLException("SELECT FOR UPDATE query requires transactional " +
- "cache with MVCC enabled.", IgniteQueryErrorCode.UNSUPPORTED_OPERATION);
-
- qry = GridSqlQueryParser.rewriteQueryForUpdateIfNeeded(select.statement(), forUpdate = tx != null);
-
- if (forUpdate) {
- GridCacheContext cctx = mvccTracker.context();
-
- try {
- if (tx.topologyVersionSnapshot() == null)
- new TxTopologyVersionFuture(tx, cctx).get();
- }
- catch (Exception e) {
- throw new IgniteSQLException("Failed to lock topology for SELECT FOR UPDATE query.", e);
- }
-
- sfuFut = new GridNearTxSelectForUpdateFuture(cctx, tx, opTimeout);
+ if (ctx.security().enabled())
+ checkSecurity(select.cacheIds());
- sfuFut.initLocal();
- }
- }
- }
+ MvccSnapshot mvccSnapshot = null;
- GridNearTxLocal tx0 = tx;
- MvccQueryTracker mvccTracker0 = mvccTracker;
- GridNearTxSelectForUpdateFuture sfuFut0 = sfuFut;
- String qry0 = qry;
- int timeout0 = opTimeout;
+ if (select.mvccEnabled())
+ mvccSnapshot = mvccTracker.snapshot();
final QueryContext qctx = new QueryContext(
0,
@@ -547,7 +496,7 @@ public class IgniteH2Indexing implements GridQueryIndexing {
PreparedStatement stmt = preparedStatementWithParams(
conn0,
- qry0,
+ qry,
args,
true
);
@@ -555,73 +504,22 @@ public class IgniteH2Indexing implements GridQueryIndexing {
ResultSet rs = executeSqlQueryWithTimer(
stmt,
conn0,
- qry0,
+ qry,
args,
- timeout0,
+ timeout,
cancel,
qryParams.dataPageScanEnabled(),
- new H2QueryInfo(H2QueryInfo.QueryType.LOCAL, stmt, qry0)
+ new H2QueryInfo(H2QueryInfo.QueryType.LOCAL, stmt, qry)
);
- if (sfuFut0 != null) {
- assert tx0.mvccSnapshot() != null;
-
- ResultSetEnlistFuture enlistFut = ResultSetEnlistFuture.future(
- ctx.localNodeId(),
- tx0.nearXidVersion(),
- tx0.mvccSnapshot(),
- tx0.threadId(),
- IgniteUuid.randomUuid(),
- -1,
- null,
- tx0,
- timeout0,
- sfuFut0.cache(),
- rs
- );
-
- enlistFut.listen(new IgniteInClosure<IgniteInternalFuture<Long>>() {
- @Override public void apply(IgniteInternalFuture<Long> fut) {
- if (fut.error() != null) {
- sfuFut0.onResult(
- ctx.localNodeId(),
- 0L,
- false,
- fut.error());
- }
- else {
- sfuFut0.onResult(
- ctx.localNodeId(),
- fut.result(),
- false,
- null);
- }
- }
- });
-
- enlistFut.init();
-
- try {
- sfuFut0.get();
-
- rs.beforeFirst();
- }
- catch (Exception e) {
- U.closeQuiet(rs);
-
- throw new IgniteSQLException("Failed to obtain locks on result of SELECT FOR UPDATE.",
- e);
- }
- }
-
- return new H2FieldsIterator(rs, mvccTracker0, sfuFut0 != null, conn);
+ return new H2FieldsIterator(rs, mvccTracker, conn);
}
catch (IgniteCheckedException | RuntimeException | Error e) {
conn.recycle();
try {
- if (mvccTracker0 != null)
- mvccTracker0.onDone();
+ if (mvccTracker != null)
+ mvccTracker.onDone();
}
catch (Exception e0) {
e.addSuppressed(e0);
@@ -635,7 +533,9 @@ public class IgniteH2Indexing implements GridQueryIndexing {
}
};
}
- catch (IgniteCheckedException | RuntimeException | Error e) {
+ catch (Exception e) {
+ GridNearTxLocal tx = null;
+
if (mvccEnabled && (tx != null || (tx = tx(ctx)) != null))
tx.setRollbackOnly();
@@ -952,94 +852,6 @@ public class IgniteH2Indexing implements GridQueryIndexing {
}
}
- /**
- * @param schemaName Schema name.
- * @param qry Query.
- * @param keepCacheObj Flag to keep cache object.
- * @param enforceJoinOrder Enforce join order of tables.
- * @param autoStartTx Start transaction flag.
- * @param qryTimeout Query timeout.
- * @param cancel Cancel object.
- * @param params Query parameters.
- * @param parts Partitions.
- * @param lazy Lazy query execution flag.
- * @param mvccTracker Query tracker.
- * @param dataPageScanEnabled If data page scan is enabled.
- * @param pageSize Page size.
- * @return Iterable result.
- */
- @SuppressWarnings("IfMayBeConditional")
- private Iterable<List<?>> runQueryTwoStep(
- final String schemaName,
- final GridCacheTwoStepQuery qry,
- final boolean keepCacheObj,
- final boolean enforceJoinOrder,
- boolean autoStartTx,
- final int qryTimeout,
- final GridQueryCancel cancel,
- final Object[] params,
- final int[] parts,
- final boolean lazy,
- MvccQueryTracker mvccTracker,
- Boolean dataPageScanEnabled,
- int pageSize
- ) {
- assert !qry.mvccEnabled() || !F.isEmpty(qry.cacheIds());
-
- try {
- final MvccQueryTracker mvccTracker0;
-
- if (mvccTracker == null && qry.mvccEnabled()) {
- mvccTracker0 = MvccUtils.mvccTracker(
- ctx.cache().context().cacheContext(qry.cacheIds().get(0)),
- autoStartTx
- );
- }
- else
- mvccTracker0 = mvccTracker;
-
- GridNearTxLocal tx = mvccTracker0 != null ? tx(ctx) : null;
-
- // Locking has no meaning if SELECT FOR UPDATE is not executed in explicit transaction.
- // So, we can can reset forUpdate flag if there is no explicit transaction.
- boolean forUpdate = qry.forUpdate() && checkActive(tx) != null;
-
- int opTimeout = operationTimeout(qryTimeout, tx);
-
- return new Iterable<List<?>>() {
- @SuppressWarnings("NullableProblems")
- @Override public Iterator<List<?>> iterator() {
- try {
- return rdcQryExec.query(
- schemaName,
- qry,
- keepCacheObj,
- enforceJoinOrder,
- opTimeout,
- cancel,
- params,
- parts,
- lazy,
- mvccTracker0,
- dataPageScanEnabled,
- forUpdate,
- pageSize
- );
- }
- catch (Throwable e) {
- if (mvccTracker0 != null)
- mvccTracker0.onDone();
-
- throw e;
- }
- }
- };
- }
- catch (IgniteCheckedException e) {
- throw U.convertException(e);
- }
- }
-
/** {@inheritDoc} */
@SuppressWarnings("deprecation")
@Override public SqlFieldsQuery generateFieldsQuery(String cacheName, SqlQuery qry) {
@@ -1351,7 +1163,49 @@ public class IgniteH2Indexing implements GridQueryIndexing {
Long qryId = registerRunningQuery(qryDesc, cancel);
try {
- Iterable<List<?>> iter = executeSelect0(qryDesc, qryParams, select, keepBinary, null, cancel);
+ GridNearTxLocal tx = null;
+ MvccQueryTracker tracker = null;
+ GridCacheContext mvccCctx = null;
+
+ boolean inTx = false;
+
+ if (select.mvccEnabled()) {
+ mvccCctx = ctx.cache().context().cacheContext(select.mvccCacheId());
+
+ if (mvccCctx == null)
+ throw new IgniteCheckedException("Cache has been stopped concurrently [cacheId=" +
+ select.mvccCacheId() + ']');
+
+ boolean autoStartTx = !qryParams.autoCommit() && tx(ctx) == null;
+
+ // Start new user tx in case of autocommit == false.
+ if (autoStartTx)
+ txStart(ctx, qryParams.timeout());
+
+ tx = tx(ctx);
+
+ checkActive(tx);
+
+ inTx = tx != null;
+
+ tracker = MvccUtils.mvccTracker(mvccCctx, tx);
+ }
+
+ int timeout = operationTimeout(qryParams.timeout(), tx);
+
+ Iterable<List<?>> iter = executeSelect0(
+ qryDesc,
+ qryParams,
+ select,
+ keepBinary,
+ tracker,
+ cancel,
+ inTx,
+ timeout);
+
+ // Execute SELECT FOR UPDATE if needed.
+ if (select.forUpdate() && inTx)
+ iter = lockSelectedRows(iter, mvccCctx, timeout, qryParams.pageSize());
QueryCursorImpl<List<?>> cursor = qryId != null
? new RegisteredQueryCursor<>(iter, cancel, runningQueryManager(), qryId)
@@ -1364,6 +1218,12 @@ public class IgniteH2Indexing implements GridQueryIndexing {
catch (Exception e) {
runningQryMgr.unregister(qryId, true);
+ if (e instanceof IgniteCheckedException)
+ throw U.convertException((IgniteCheckedException)e);
+
+ if (e instanceof RuntimeException)
+ throw (RuntimeException)e;
+
throw new IgniteSQLException("Failed to execute SELECT statement: " + qryDesc.sql(), e);
}
}
@@ -1375,13 +1235,15 @@ public class IgniteH2Indexing implements GridQueryIndexing {
* @param selectQry Select query.
* @param mvccTracker MVCC tracker.
* @param cancel Cancel.
+ * @param timeout Timeout.
* @return Fields query.
*/
private QueryCursorImpl<List<?>> executeSelectForDml(
String schema,
SqlFieldsQuery selectQry,
MvccQueryTracker mvccTracker,
- GridQueryCancel cancel
+ GridQueryCancel cancel,
+ int timeout
) throws IgniteCheckedException {
QueryParserResult parseRes = parser.parse(schema, selectQry, false);
@@ -1395,7 +1257,9 @@ public class IgniteH2Indexing implements GridQueryIndexing {
select,
true,
mvccTracker,
- cancel
+ cancel,
+ false,
+ timeout
);
QueryCursorImpl<List<?>> cursor = new QueryCursorImpl<>(iter, cancel);
@@ -1414,6 +1278,8 @@ public class IgniteH2Indexing implements GridQueryIndexing {
* @param keepBinary Whether binary objects must not be deserialized automatically.
* @param mvccTracker MVCC tracker.
* @param cancel Query cancel state holder.
+ * @param inTx Flag whether query is executed within transaction.
+ * @param timeout Timeout.
* @return Query result.
*/
private Iterable<List<?>> executeSelect0(
@@ -1422,18 +1288,18 @@ public class IgniteH2Indexing implements GridQueryIndexing {
QueryParserResultSelect select,
boolean keepBinary,
MvccQueryTracker mvccTracker,
- GridQueryCancel cancel
+ GridQueryCancel cancel,
+ boolean inTx,
+ int timeout
) throws IgniteCheckedException {
- if (ctx.security().enabled())
- checkSecurity(select.cacheIds());
-
- boolean autoStartTx = mvccEnabled(ctx) && !qryParams.autoCommit() && tx(ctx) == null;
+ assert !select.mvccEnabled() || mvccTracker != null;
Iterable<List<?>> iter;
if (select.splitNeeded()) {
// Distributed query.
- GridCacheTwoStepQuery twoStepQry = select.twoStepQuery();
+ GridCacheTwoStepQuery twoStepQry = select.forUpdate() && inTx ?
+ select.forUpdateTwoStepQuery() : select.twoStepQuery();
assert twoStepQry != null;
@@ -1442,9 +1308,9 @@ public class IgniteH2Indexing implements GridQueryIndexing {
qryParams,
twoStepQry,
keepBinary,
- autoStartTx,
mvccTracker,
- cancel
+ cancel,
+ timeout
);
}
else {
@@ -1456,9 +1322,10 @@ public class IgniteH2Indexing implements GridQueryIndexing {
qryParams,
select,
filter,
- autoStartTx,
mvccTracker,
- cancel
+ cancel,
+ inTx,
+ timeout
);
iter = () -> {
@@ -1475,6 +1342,69 @@ public class IgniteH2Indexing implements GridQueryIndexing {
}
/**
+ * Locks rows from query cursor and returns the select result.
+ *
+ * @param cur Query cursor.
+ * @param cctx Cache context.
+ * @param pageSize Page size.
+ * @param timeout Timeout.
+ * @return Query results cursor.
+ */
+ private Iterable<List<?>> lockSelectedRows(Iterable<List<?>> cur, GridCacheContext cctx, int pageSize, long timeout) {
+ assert cctx != null && cctx.mvccEnabled();
+
+ GridNearTxLocal tx = tx(ctx);
+
+ if (tx == null)
+ throw new IgniteSQLException("Failed to perform SELECT FOR UPDATE operation: transaction has already finished.");
+
+ Collection<List<?>> rowsCache = new ArrayList<>();
+
+ UpdateSourceIterator srcIt = new UpdateSourceIterator<KeyCacheObject>() {
+ private Iterator<List<?>> it = cur.iterator();
+
+ @Override public EnlistOperation operation() {
+ return EnlistOperation.LOCK;
+ }
+
+ @Override public boolean hasNextX() throws IgniteCheckedException {
+ return it.hasNext();
+ }
+
+ @Override public KeyCacheObject nextX() throws IgniteCheckedException {
+ List<?> res = it.next();
+
+ // nextX() can be called from the different threads.
+ synchronized (rowsCache) {
+ rowsCache.add(res.subList(0, res.size() - 1));
+
+ if (rowsCache.size() > TX_SIZE_THRESHOLD) {
+ throw new IgniteCheckedException("Too many rows are locked by SELECT FOR UPDATE statement. " +
+ "Consider locking fewer keys or increase the limit by setting a " +
+ IGNITE_MVCC_TX_SIZE_CACHING_THRESHOLD + " system property. Current value is " +
+ TX_SIZE_THRESHOLD + " rows.");
+ }
+ }
+
+ // The last column is expected to be a _key.
+ return cctx.toCacheKeyObject(res.get(res.size() - 1));
+ }
+ };
+
+ IgniteInternalFuture<Long> fut = tx.updateAsync(cctx, srcIt, pageSize,
+ timeout, true);
+
+ try {
+ fut.get();
+ }
+ catch (IgniteCheckedException e) {
+ throw U.convertException(e);
+ }
+
+ return rowsCache;
+ }
+
+ /**
* Register running query.
*
* @param qryDesc Query descriptor.
@@ -1584,7 +1514,8 @@ public class IgniteH2Indexing implements GridQueryIndexing {
schema,
selectFieldsQry,
new StaticMvccQueryTracker(planCctx, mvccSnapshot),
- cancel
+ cancel,
+ timeout
);
}
else {
@@ -1597,9 +1528,10 @@ public class IgniteH2Indexing implements GridQueryIndexing {
selectParseRes.queryParameters(),
selectParseRes.select(),
filter,
- false,
new StaticMvccQueryTracker(planCctx, mvccSnapshot),
- cancel
+ cancel,
+ true,
+ timeout
);
cur = new QueryCursorImpl<>(new Iterable<List<?>>() {
@@ -1625,25 +1557,25 @@ public class IgniteH2Indexing implements GridQueryIndexing {
* @param qryParams Query parameters.
* @param twoStepQry Two-step query.
* @param keepBinary Keep binary flag.
- * @param autoStartTx Start transaction flag.
* @param mvccTracker Query tracker.
* @param cancel Cancel handler.
+ * @param timeout Timeout.
* @return Cursor representing distributed query result.
*/
@SuppressWarnings("IfMayBeConditional")
private Iterable<List<?>> executeSelectDistributed(
- QueryDescriptor qryDesc,
- QueryParameters qryParams,
- GridCacheTwoStepQuery twoStepQry,
- boolean keepBinary,
- boolean autoStartTx,
+ final QueryDescriptor qryDesc,
+ final QueryParameters qryParams,
+ final GridCacheTwoStepQuery twoStepQry,
+ final boolean keepBinary,
MvccQueryTracker mvccTracker,
- GridQueryCancel cancel
+ final GridQueryCancel cancel,
+ int timeout
) {
// When explicit partitions are set, there must be an owning cache they should be applied to.
PartitionResult derivedParts = twoStepQry.derivedPartitions();
- int parts[] = PartitionResult.calculatePartitions(
+ final int parts[] = PartitionResult.calculatePartitions(
qryParams.partitions(),
derivedParts,
qryParams.arguments()
@@ -1669,21 +1601,36 @@ public class IgniteH2Indexing implements GridQueryIndexing {
};
}
else {
- iter = runQueryTwoStep(
- qryDesc.schemaName(),
- twoStepQry,
- keepBinary,
- qryDesc.enforceJoinOrder(),
- autoStartTx,
- qryParams.timeout(),
- cancel,
- qryParams.arguments(),
- parts,
- qryParams.lazy(),
- mvccTracker,
- qryParams.dataPageScanEnabled(),
- qryParams.pageSize()
- );
+ assert !twoStepQry.mvccEnabled() || !F.isEmpty(twoStepQry.cacheIds());
+ assert twoStepQry.mvccEnabled() == (mvccTracker != null);
+
+ iter = new Iterable<List<?>>() {
+ @SuppressWarnings("NullableProblems")
+ @Override public Iterator<List<?>> iterator() {
+ try {
+ return IgniteH2Indexing.this.rdcQryExec.query(
+ qryDesc.schemaName(),
+ twoStepQry,
+ keepBinary,
+ qryDesc.enforceJoinOrder(),
+ timeout,
+ cancel,
+ qryParams.arguments(),
+ parts,
+ qryParams.lazy(),
+ mvccTracker,
+ qryParams.dataPageScanEnabled(),
+ qryParams.pageSize()
+ );
+ }
+ catch (Throwable e) {
+ if (mvccTracker != null)
+ mvccTracker.onDone();
+
+ throw e;
+ }
+ }
+ };
}
return iter;
@@ -2544,7 +2491,8 @@ public class IgniteH2Indexing implements GridQueryIndexing {
qryDesc.schemaName(),
selectFieldsQry,
null,
- cancel
+ cancel,
+ qryParams.timeout()
);
}
else if (plan.hasRows())
@@ -2559,9 +2507,10 @@ public class IgniteH2Indexing implements GridQueryIndexing {
selectParseRes.queryParameters(),
selectParseRes.select(),
filters,
- false,
null,
- cancel
+ cancel,
+ false,
+ qryParams.timeout()
);
cur = new QueryCursorImpl<>(new Iterable<List<?>>() {
@@ -2662,7 +2611,8 @@ public class IgniteH2Indexing implements GridQueryIndexing {
qryDesc.schemaName(),
selectFieldsQry,
MvccUtils.mvccTracker(cctx, tx),
- cancel
+ cancel,
+ (int)timeout
);
it = plan.iteratorForTransaction(connMgr, cur);
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/NearResultSetEnlistFuture.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/NearResultSetEnlistFuture.java
deleted file mode 100644
index 968a856..0000000
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/NearResultSetEnlistFuture.java
+++ /dev/null
@@ -1,59 +0,0 @@
-/*
- * 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.h2;
-
-import java.sql.ResultSet;
-import java.util.UUID;
-import org.apache.ignite.internal.processors.cache.GridCacheContext;
-import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxLocalAdapter;
-import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
-import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxQueryResultsEnlistFuture;
-import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
-import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
-import org.apache.ignite.lang.IgniteUuid;
-import org.jetbrains.annotations.Nullable;
-
-/**
- *
- */
-public class NearResultSetEnlistFuture extends GridNearTxQueryResultsEnlistFuture implements ResultSetEnlistFuture {
- /**
- * @param nearNodeId Near node ID.
- * @param nearLockVer Near lock version.
- * @param mvccSnapshot Mvcc snapshot.
- * @param threadId Thread ID.
- * @param nearFutId Near future id.
- * @param nearMiniId Near mini future id.
- * @param parts Partitions.
- * @param tx Transaction.
- * @param timeout Lock acquisition timeout.
- * @param cctx Cache context.
- * @param rs Result set to process.
- */
- public NearResultSetEnlistFuture(UUID nearNodeId, GridCacheVersion nearLockVer, MvccSnapshot mvccSnapshot,
- long threadId, IgniteUuid nearFutId, int nearMiniId, @Nullable int[] parts, GridDhtTxLocalAdapter tx,
- long timeout, GridCacheContext<?, ?> cctx, ResultSet rs) {
- super(cctx, (GridNearTxLocal)tx, timeout, ResultSetEnlistFuture.createIterator(rs), 0, true);
- }
-
-
- /** {@inheritDoc} */
- @Override public void onError(Throwable error) {
- onDone(error);
- }
-}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/QueryParser.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/QueryParser.java
index 7f6a720..9f12ec9 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/QueryParser.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/QueryParser.java
@@ -45,8 +45,10 @@ import org.apache.ignite.internal.processors.query.h2.opt.GridH2Table;
import org.apache.ignite.internal.processors.query.h2.sql.GridSqlAlias;
import org.apache.ignite.internal.processors.query.h2.sql.GridSqlAst;
import org.apache.ignite.internal.processors.query.h2.sql.GridSqlInsert;
+import org.apache.ignite.internal.processors.query.h2.sql.GridSqlQuery;
import org.apache.ignite.internal.processors.query.h2.sql.GridSqlQueryParser;
import org.apache.ignite.internal.processors.query.h2.sql.GridSqlQuerySplitter;
+import org.apache.ignite.internal.processors.query.h2.sql.GridSqlSelect;
import org.apache.ignite.internal.processors.query.h2.sql.GridSqlStatement;
import org.apache.ignite.internal.processors.query.h2.sql.GridSqlTable;
import org.apache.ignite.internal.sql.SqlParseException;
@@ -70,6 +72,8 @@ import org.apache.ignite.internal.util.typedef.internal.U;
import org.h2.command.Prepared;
import org.jetbrains.annotations.Nullable;
+import static org.apache.ignite.internal.processors.query.h2.sql.GridSqlQuerySplitter.keyColumn;
+
/**
* Parser module. Splits incoming request into a series of parsed results.
*/
@@ -114,7 +118,7 @@ public class QueryParser {
*
* @param schemaName schema name.
* @param qry query to parse.
- * @param remainingAllowed Whether multiple statements are allowed.
+ * @param remainingAllowed Whether multiple statements are allowed.
* @return Parsing result that contains Parsed leading query and remaining sql script.
*/
public QueryParserResult parse(String schemaName, SqlFieldsQuery qry, boolean remainingAllowed) {
@@ -138,7 +142,7 @@ public class QueryParser {
QueryParserCacheEntry cached = cache.get(qryDesc);
- if (cached != null)
+ if (cached != null)
return new QueryParserResult(
qryDesc,
QueryParameters.fromQuery(qry),
@@ -306,7 +310,7 @@ public class QueryParser {
// Prepare new query.
SqlFieldsQuery newQry = cloneFieldsQuery(qry).setSql(prepared.getSQL());
- int paramsCnt = prepared.getParameters().size();
+ final int paramsCnt = prepared.getParameters().size();
Object[] argsOrig = qry.getArgs();
@@ -381,7 +385,7 @@ public class QueryParser {
// Parse SELECT.
GridSqlQueryParser parser = new GridSqlQueryParser(false);
- GridSqlStatement stmt0 = parser.parse(prepared);
+ GridSqlQuery selectStmt = (GridSqlQuery)parser.parse(prepared);
List<Integer> cacheIds = parser.cacheIds();
Integer mvccCacheId = mvccCacheIdForSelect(parser.objectsMap());
@@ -410,32 +414,88 @@ public class QueryParser {
boolean splitNeeded = !loc || locSplit;
try {
+ String forUpdateQryOutTx = null;
+ String forUpdateQryTx = null;
+ GridCacheTwoStepQuery forUpdateTwoStepQry = null;
+
+ boolean forUpdate = GridSqlQueryParser.isForUpdateQuery(prepared);
+
+ // SELECT FOR UPDATE case handling. We need to create extra queries with appended _key
+ // column to be able to lock selected rows further.
+ if (forUpdate) {
+ // We have checked above that it's not an UNION query, so it's got to be SELECT.
+ assert selectStmt instanceof GridSqlSelect;
+
+ // Check FOR UPDATE invariants: only one table, MVCC is there.
+ if (cacheIds.size() != 1)
+ throw new IgniteSQLException("SELECT FOR UPDATE is supported only for queries " +
+ "that involve single transactional cache.");
+
+ if (mvccCacheId == null)
+ throw new IgniteSQLException("SELECT FOR UPDATE query requires transactional cache " +
+ "with MVCC enabled.", IgniteQueryErrorCode.UNSUPPORTED_OPERATION);
+
+ // We need a copy because we are going to modify AST a bit. We do not want to modify original select.
+ GridSqlSelect selForUpdate = ((GridSqlSelect)selectStmt).copySelectForUpdate();
+
+ // Clear forUpdate flag to run it as a plain query.
+ selForUpdate.forUpdate(false);
+ ((GridSqlSelect)selectStmt).forUpdate(false);
+
+ // Remember sql string without FOR UPDATE clause.
+ forUpdateQryOutTx = selForUpdate.getSQL();
+
+ GridSqlAlias keyCol = keyColumn(selForUpdate);
+
+ selForUpdate.addColumn(keyCol, true);
+
+ // Remember sql string without FOR UPDATE clause and with _key column.
+ forUpdateQryTx = selForUpdate.getSQL();
+
+ // Prepare additional two-step query for FOR UPDATE case.
+ if (splitNeeded) {
+ forUpdateTwoStepQry = GridSqlQuerySplitter.split(
+ connMgr.connectionForThread().connection(newQry.getSchema()),
+ selForUpdate,
+ forUpdateQryTx,
+ newQry.isCollocated(),
+ newQry.isDistributedJoins(),
+ newQry.isEnforceJoinOrder(),
+ locSplit,
+ idx,
+ paramsCnt
+ );
+ }
+ }
+
GridCacheTwoStepQuery twoStepQry = null;
if (splitNeeded) {
twoStepQry = GridSqlQuerySplitter.split(
connMgr.connectionForThread().connection(newQry.getSchema()),
- prepared,
+ selectStmt,
+ newQry.getSql(),
newQry.isCollocated(),
newQry.isDistributedJoins(),
newQry.isEnforceJoinOrder(),
locSplit,
- idx
+ idx,
+ paramsCnt
);
}
List<GridQueryFieldMetadata> meta = H2Utils.meta(stmt.getMetaData());
- boolean forUpdate = GridSqlQueryParser.isForUpdateQuery(prepared);
-
QueryParserResultSelect select = new QueryParserResultSelect(
- stmt0,
+ selectStmt,
twoStepQry,
+ forUpdateTwoStepQry,
meta,
- prepared.getParameters().size(),
+ paramsCnt,
cacheIds,
mvccCacheId,
- forUpdate
+ forUpdateQryOutTx,
+ forUpdateQryTx
);
return new QueryParserResult(
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/QueryParserResultSelect.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/QueryParserResultSelect.java
index 1c4be7d..eee7c81 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/QueryParserResultSelect.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/QueryParserResultSelect.java
@@ -29,12 +29,23 @@ import java.util.List;
*/
@SuppressWarnings("AssignmentOrReturnOfFieldWithMutableType")
public class QueryParserResultSelect {
- /** Statmement. */
+ /** Statement. */
private GridSqlStatement stmt;
/** Two-step query, or {@code} null if this result is for local query. */
private final GridCacheTwoStepQuery twoStepQry;
+ /**
+ * Two-step query in SELECT FOR UPDATE case, or {@code} null if this result is for local query.
+ * If a query is for update, we need to save two variants of the this query.
+ * First variant {@link QueryParserResultSelect#twoStepQry} is used when the query is executed outside
+ * of transaction - it is executed as a plain query. The second variant of the query - is actually
+ * a "for update" query which is used when running within transaction. In this query an extra _key column
+ * is implicitly appended to query columns. This extra column is used to lock the selected rows.
+ * This column is hidden from client.
+ */
+ private final GridCacheTwoStepQuery forUpdateTwoStepQry;
+
/** Metadata for two-step query, or {@code} null if this result is for local query. */
private final List<GridQueryFieldMetadata> meta;
@@ -47,36 +58,51 @@ public class QueryParserResultSelect {
/** ID of the first MVCC cache. */
private final Integer mvccCacheId;
- /** FOR UPDATE flag. */
- private final boolean forUpdate;
+ /**
+ * Sql query with cleared "FOR UPDATE" statement.
+ * This string is used when query is executed out of transaction.
+ */
+ private final String forUpdateQryOutTx;
+
+ /**
+ * Sql query for update. Contains additional "_key" column.
+ * This string is used when executing query within explicit transaction.
+ */
+ private final String forUpdateQryTx;
/**
* Constructor.
*
* @param stmt Statement.
* @param twoStepQry Distributed query plan.
+ * @param forUpdateTwoStepQry FOR UPDATE query for execution within transaction.
* @param meta Fields metadata.
* @param paramsCnt Parameters count.
* @param cacheIds Cache IDs.
* @param mvccCacheId ID of the first MVCC cache.
- * @param forUpdate Whether this is FOR UPDATE flag.
+ * @param forUpdateQryOutTx FOR UPDATE query string for execution out of transaction.
+ * @param forUpdateQryTx FOR UPDATE query string for execution within transaction.
*/
public QueryParserResultSelect(
GridSqlStatement stmt,
@Nullable GridCacheTwoStepQuery twoStepQry,
+ @Nullable GridCacheTwoStepQuery forUpdateTwoStepQry,
List<GridQueryFieldMetadata> meta,
int paramsCnt,
List<Integer> cacheIds,
@Nullable Integer mvccCacheId,
- boolean forUpdate
+ String forUpdateQryOutTx,
+ String forUpdateQryTx
) {
this.stmt = stmt;
this.twoStepQry = twoStepQry;
+ this.forUpdateTwoStepQry = forUpdateTwoStepQry;
this.meta = meta;
this.paramsCnt = paramsCnt;
this.cacheIds = cacheIds;
this.mvccCacheId = mvccCacheId;
- this.forUpdate = forUpdate;
+ this.forUpdateQryOutTx = forUpdateQryOutTx;
+ this.forUpdateQryTx = forUpdateQryTx;
}
/**
@@ -94,6 +120,13 @@ public class QueryParserResultSelect {
}
/**
+ * @return Two-step query for update, or {@code} null if this result is for local query.
+ */
+ @Nullable public GridCacheTwoStepQuery forUpdateTwoStepQuery() {
+ return forUpdateTwoStepQry;
+ }
+
+ /**
* @return Two-step query metadata.
*/
public List<GridQueryFieldMetadata> meta() {
@@ -132,7 +165,21 @@ public class QueryParserResultSelect {
* @return Whether this is FOR UPDATE query.
*/
public boolean forUpdate() {
- return forUpdate;
+ return forUpdateQryTx != null;
+ }
+
+ /**
+ * @return Sql FOR UPDATE query for execution out of transaction.
+ */
+ public String forUpdateQueryOutTx() {
+ return forUpdateQryOutTx;
+ }
+
+ /**
+ * @return Sql FOR UPDATE query for execution within transaction.
+ */
+ public String forUpdateQueryTx() {
+ return forUpdateQryTx;
}
/**
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/ResultSetEnlistFuture.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/ResultSetEnlistFuture.java
deleted file mode 100644
index ee1c0fa..0000000
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/ResultSetEnlistFuture.java
+++ /dev/null
@@ -1,136 +0,0 @@
-/*
- * 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.h2;
-
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.util.NoSuchElementException;
-import java.util.UUID;
-import org.apache.ignite.internal.processors.cache.GridCacheContext;
-import org.apache.ignite.internal.processors.cache.distributed.dht.DhtLockFuture;
-import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxLocalAdapter;
-import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
-import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
-import org.apache.ignite.internal.processors.query.EnlistOperation;
-import org.apache.ignite.internal.processors.query.IgniteSQLException;
-import org.apache.ignite.internal.processors.query.UpdateSourceIterator;
-import org.apache.ignite.lang.IgniteUuid;
-import org.jetbrains.annotations.Nullable;
-
-/**
- * Future to process whole local result set of SELECT FOR UPDATE query.
- */
-public interface ResultSetEnlistFuture extends DhtLockFuture<Long> {
- /**
- * @param rs Result set.
- * @return Update source.
- */
- static UpdateSourceIterator<?> createIterator(ResultSet rs) {
- return new ResultSetUpdateSourceIteratorWrapper(rs);
- }
-
- /** */
- void init();
-
- /**
- *
- * @param nearNodeId Near node ID.
- * @param nearLockVer Near lock version.
- * @param mvccSnapshot Mvcc snapshot.
- * @param threadId Thread ID.
- * @param nearFutId Near future id.
- * @param nearMiniId Near mini future id.
- * @param parts Partitions.
- * @param tx Transaction.
- * @param timeout Lock acquisition timeout.
- * @param cctx Cache context.
- * @param rs Result set to process.
- * @return Result set enlist future.
- */
- static ResultSetEnlistFuture future(UUID nearNodeId, GridCacheVersion nearLockVer,
- MvccSnapshot mvccSnapshot, long threadId, IgniteUuid nearFutId, int nearMiniId, @Nullable int[] parts,
- GridDhtTxLocalAdapter tx, long timeout, GridCacheContext<?, ?> cctx, ResultSet rs) {
-
- if (tx.near())
- return new NearResultSetEnlistFuture(nearNodeId, nearLockVer, mvccSnapshot, threadId, nearFutId, nearMiniId, parts, tx, timeout, cctx, rs);
- else
- return new DhtResultSetEnlistFuture(nearNodeId, nearLockVer, mvccSnapshot, threadId, nearFutId, nearMiniId, parts, tx, timeout, cctx, rs);
- }
-
- /**
- *
- */
- public static class ResultSetUpdateSourceIteratorWrapper implements UpdateSourceIterator<Object> {
- /** */
- private static final long serialVersionUID = -8745196216234843471L;
-
- /** */
- private final ResultSet rs;
-
- /** */
- private Boolean hasNext;
-
- /** */
- private int keyColIdx;
-
- /**
- * @param rs Result set.
- */
- public ResultSetUpdateSourceIteratorWrapper(ResultSet rs) {
- this.rs = rs;
- keyColIdx = -1;
- }
-
- /** {@inheritDoc} */
- @Override public EnlistOperation operation() {
- return EnlistOperation.LOCK;
- }
-
- /** {@inheritDoc} */
- @Override public boolean hasNextX() {
- try {
- if (hasNext == null)
- hasNext = rs.next();
-
- return hasNext;
- }
- catch (SQLException e) {
- throw new IgniteSQLException(e);
- }
- }
-
- /** {@inheritDoc} */
- @Override public Object nextX() {
- if (!hasNextX())
- throw new NoSuchElementException();
-
- try {
- if (keyColIdx == -1)
- keyColIdx = rs.getMetaData().getColumnCount();
-
- return rs.getObject(keyColIdx);
- }
- catch (SQLException e) {
- throw new IgniteSQLException(e);
- }
- finally {
- hasNext = null;
- }
- }
- }
-}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/dml/UpdatePlanBuilder.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/dml/UpdatePlanBuilder.java
index 033a892..8084743 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/dml/UpdatePlanBuilder.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/dml/UpdatePlanBuilder.java
@@ -66,6 +66,7 @@ import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteClosure;
+import org.h2.command.Prepared;
import org.h2.table.Column;
/**
@@ -892,14 +893,19 @@ public final class UpdatePlanBuilder {
try (Connection conn = idx.connections().connectionNoCache(planKey.schemaName())) {
// Get a new prepared statement for derived select query.
try (PreparedStatement stmt = conn.prepareStatement(selectQry)) {
+ Prepared prep = GridSqlQueryParser.prepared(stmt);
+ GridSqlQuery selectStmt = (GridSqlQuery)new GridSqlQueryParser(false).parse(prep);
+
GridCacheTwoStepQuery qry = GridSqlQuerySplitter.split(
conn,
- GridSqlQueryParser.prepared(stmt),
+ selectStmt,
+ selectQry,
planKey.collocated(),
planKey.distributedJoins(),
planKey.enforceJoinOrder(),
false,
- idx
+ idx,
+ prep.getParameters().size()
);
boolean distributed =
@@ -911,7 +917,7 @@ public final class UpdatePlanBuilder {
if (distributed) {
List<Integer> cacheIds = H2Utils.collectCacheIds(idx, CU.cacheId(cacheName), qry.tables());
- H2Utils.checkQuery(idx, cacheIds, qry.mvccEnabled(), qry.forUpdate(), qry.tables());
+ H2Utils.checkQuery(idx, cacheIds, qry.tables());
return new DmlDistributedPlanInfo(qry.isReplicatedOnly(), cacheIds, qry.derivedPartitions());
}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sql/GridSqlQueryParser.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sql/GridSqlQueryParser.java
index 48ccb84..fb40780 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sql/GridSqlQueryParser.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sql/GridSqlQueryParser.java
@@ -42,7 +42,6 @@ import org.apache.ignite.internal.processors.query.h2.dml.DmlAstUtils;
import org.apache.ignite.internal.processors.query.h2.opt.GridH2RowDescriptor;
import org.apache.ignite.internal.processors.query.h2.opt.GridH2Table;
import org.apache.ignite.internal.util.typedef.F;
-import org.apache.ignite.lang.IgniteUuid;
import org.h2.command.Command;
import org.h2.command.CommandContainer;
import org.h2.command.CommandInterface;
@@ -618,52 +617,6 @@ public class GridSqlQueryParser {
}
/**
- * @param p Statement to rewrite, if needed.
- * @param inTx Whether there is an active transaction.
- * @return Query with {@code key} and {@code val} columns appended to the list of columns,
- * if it's an {@code FOR UPDATE} query, or {@code null} if nothing has to be done.
- */
- @NotNull public static String rewriteQueryForUpdateIfNeeded(Prepared p, boolean inTx) {
- GridSqlStatement gridStmt = new GridSqlQueryParser(false).parse(p);
- return rewriteQueryForUpdateIfNeeded(gridStmt, inTx);
- }
-
- /**
- * @param stmt Statement to rewrite, if needed.
- * @param inTx Whether there is an active transaction.
- * @return Query with {@code key} and {@code val} columns appended to the list of columns,
- * if it's an {@code FOR UPDATE} query, or {@code null} if nothing has to be done.
- */
- @NotNull public static String rewriteQueryForUpdateIfNeeded(GridSqlStatement stmt, boolean inTx) {
- // We have checked above that it's not an UNION query, so it's got to be SELECT.
- assert stmt instanceof GridSqlSelect;
-
- GridSqlSelect sel = (GridSqlSelect)stmt;
-
- // How'd we get here otherwise?
- assert sel.isForUpdate();
-
- if (inTx) {
- GridSqlAst from = sel.from();
-
- GridSqlTable gridTbl = from instanceof GridSqlTable ? (GridSqlTable)from :
- ((GridSqlAlias)from).child();
-
- GridH2Table tbl = gridTbl.dataTable();
-
- Column keyCol = tbl.getColumn(0);
-
- sel.addColumn(new GridSqlAlias("_key_" + IgniteUuid.vmId(),
- new GridSqlColumn(keyCol, null, keyCol.getName()), true), true);
- }
-
- // We need to remove this flag for final flag we'll feed to H2.
- sel.forUpdate(false);
-
- return sel.getSQL();
- }
-
- /**
* @param qry Query expression to parse.
* @return Subquery AST.
*/
@@ -825,6 +778,14 @@ public class GridSqlQueryParser {
throw new IgniteSQLException("SELECT FOR UPDATE with aggregates and/or GROUP BY is not supported.",
IgniteQueryErrorCode.UNSUPPORTED_OPERATION);
}
+
+ if (select.isDistinct())
+ throw new IgniteSQLException("DISTINCT clause is not supported for SELECT FOR UPDATE.",
+ IgniteQueryErrorCode.UNSUPPORTED_OPERATION);
+
+ if (SplitterUtils.hasSubQueries(res))
+ throw new IgniteSQLException("Sub queries are not supported for SELECT FOR UPDATE.",
+ IgniteQueryErrorCode.UNSUPPORTED_OPERATION);
}
ArrayList<Expression> expressions = select.getExpressions();
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sql/GridSqlQuerySplitter.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sql/GridSqlQuerySplitter.java
index 582b62a..00ce647 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sql/GridSqlQuerySplitter.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sql/GridSqlQuerySplitter.java
@@ -38,13 +38,16 @@ import org.apache.ignite.internal.processors.cache.query.GridCacheTwoStepQuery;
import org.apache.ignite.internal.processors.cache.query.IgniteQueryErrorCode;
import org.apache.ignite.internal.processors.cache.query.QueryTable;
import org.apache.ignite.internal.processors.query.IgniteSQLException;
+import org.apache.ignite.internal.processors.query.QueryUtils;
import org.apache.ignite.internal.processors.query.h2.H2Utils;
import org.apache.ignite.internal.processors.query.h2.IgniteH2Indexing;
import org.apache.ignite.internal.processors.query.h2.affinity.PartitionExtractor;
+import org.apache.ignite.internal.processors.query.h2.opt.GridH2Table;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.h2.command.Prepared;
import org.h2.command.dml.Query;
+import org.h2.table.Column;
import static org.apache.ignite.internal.processors.query.h2.opt.join.CollocationModel.isCollocated;
import static org.apache.ignite.internal.processors.query.h2.sql.GridSqlConst.TRUE;
@@ -177,36 +180,42 @@ public class GridSqlQuerySplitter {
/**
* @param conn Connection.
- * @param prepared Prepared.
+ * @param qry Query.
+ * @param originalSql Original SQL query string.
* @param collocatedGrpBy Whether the query has collocated GROUP BY keys.
* @param distributedJoins If distributed joins enabled.
* @param enforceJoinOrder Enforce join order.
* @param locSplit Whether this is a split for local query.
* @param idx Indexing.
+ * @param paramsCnt Parameters count.
* @return Two step query.
* @throws SQLException If failed.
* @throws IgniteCheckedException If failed.
*/
public static GridCacheTwoStepQuery split(
Connection conn,
- Prepared prepared,
+ GridSqlQuery qry,
+ String originalSql,
boolean collocatedGrpBy,
boolean distributedJoins,
boolean enforceJoinOrder,
boolean locSplit,
- IgniteH2Indexing idx
+ IgniteH2Indexing idx,
+ int paramsCnt
) throws SQLException, IgniteCheckedException {
SplitterContext.set(distributedJoins);
try {
return split0(
conn,
- prepared,
+ qry,
+ originalSql,
collocatedGrpBy,
distributedJoins,
enforceJoinOrder,
locSplit,
- idx
+ idx,
+ paramsCnt
);
}
finally {
@@ -216,37 +225,33 @@ public class GridSqlQuerySplitter {
/**
* @param conn Connection.
- * @param prepared Prepared.
+ * @param qry Query.
+ * @param originalSql Original SQL query string.
* @param collocatedGrpBy Whether the query has collocated GROUP BY keys.
* @param distributedJoins If distributed joins enabled.
* @param enforceJoinOrder Enforce join order.
* @param locSplit Whether this is a split for local query.
* @param idx Indexing.
+ * @param paramsCnt Parameters count.
* @return Two step query.
* @throws SQLException If failed.
* @throws IgniteCheckedException If failed.
*/
- public static GridCacheTwoStepQuery split0(
+ private static GridCacheTwoStepQuery split0(
Connection conn,
- Prepared prepared,
+ GridSqlQuery qry,
+ String originalSql,
boolean collocatedGrpBy,
boolean distributedJoins,
boolean enforceJoinOrder,
boolean locSplit,
- IgniteH2Indexing idx
+ IgniteH2Indexing idx,
+ int paramsCnt
) throws SQLException, IgniteCheckedException {
- // Here we will just do initial query parsing. Do not use optimized
- // subqueries because we do not have unique FROM aliases yet.
- GridSqlQuery qry = GridSqlQueryParser.parseQuery(prepared, false);
-
- String originalSql = prepared.getSQL();
-
final boolean explain = qry.explain();
qry.explain(false);
- int paramsCnt = F.isEmpty(prepared.getParameters()) ? 0 : prepared.getParameters().size();
-
GridSqlQuerySplitter splitter = new GridSqlQuerySplitter(
paramsCnt,
collocatedGrpBy,
@@ -264,10 +269,8 @@ public class GridSqlQuerySplitter {
// the REDUCE query optimization.
qry = GridSqlQueryParser.parseQuery(prepare(conn, qry.getSQL(), false, enforceJoinOrder), true);
- boolean forUpdate = GridSqlQueryParser.isForUpdateQuery(prepared);
-
// Do the actual query split. We will update the original query AST, need to be careful.
- splitter.splitQuery(qry, forUpdate);
+ splitter.splitQuery(qry);
assert !F.isEmpty(splitter.mapSqlQrys): "map"; // We must have at least one map query.
assert splitter.rdcSqlQry != null: "rdc"; // We must have a reduce query.
@@ -294,19 +297,18 @@ public class GridSqlQuerySplitter {
List<Integer> cacheIds = H2Utils.collectCacheIds(idx, null, splitter.tbls);
boolean mvccEnabled = H2Utils.collectMvccEnabled(idx, cacheIds);
- H2Utils.checkQuery(idx, cacheIds, mvccEnabled, forUpdate, splitter.tbls);
+ H2Utils.checkQuery(idx, cacheIds, splitter.tbls);
// Setup resulting two step query and return it.
return new GridCacheTwoStepQuery(
originalSql,
- prepared.getParameters().size(),
+ paramsCnt,
splitter.tbls,
splitter.rdcSqlQry,
splitter.mapSqlQrys,
splitter.skipMergeTbl,
explain,
distributedJoins,
- forUpdate,
splitter.extractor.mergeMapQueries(splitter.mapSqlQrys),
cacheIds,
mvccEnabled,
@@ -316,9 +318,8 @@ public class GridSqlQuerySplitter {
/**
* @param qry Optimized and normalized query to split.
- * @param forUpdate {@code SELECT FOR UPDATE} flag.
*/
- private void splitQuery(GridSqlQuery qry, boolean forUpdate) throws IgniteCheckedException {
+ private void splitQuery(GridSqlQuery qry) throws IgniteCheckedException {
// Create a fake parent AST element for the query to allow replacing the query in the parent by split.
GridSqlSubquery fakeQryParent = new GridSqlSubquery(qry);
@@ -353,15 +354,6 @@ public class GridSqlQuerySplitter {
// Get back the updated query from the fake parent. It will be our reduce query.
qry = fakeQryParent.subquery();
- // Reset SELECT FOR UPDATE flag for reduce query.
- if (forUpdate) {
- assert qry instanceof GridSqlSelect;
-
- GridSqlSelect sel = (GridSqlSelect)qry;
-
- sel.forUpdate(false);
- }
-
String rdcQry = qry.getSQL();
SplitterUtils.checkNoDataTablesInReduceQuery(qry, rdcQry);
@@ -1268,6 +1260,30 @@ public class GridSqlQuerySplitter {
}
/**
+ * Retrieves _KEY column from SELECT. This column is used for SELECT FOR UPDATE statements.
+ *
+ * @param sel Select statement.
+ * @return Key column alias.
+ */
+ public static GridSqlAlias keyColumn(GridSqlSelect sel) {
+ GridSqlAst from = sel.from();
+
+ GridSqlTable tbl = from instanceof GridSqlTable ? (GridSqlTable)from :
+ ((GridSqlElement)from).child();
+
+ GridH2Table gridTbl = tbl.dataTable();
+
+ Column h2KeyCol = gridTbl.getColumn(QueryUtils.KEY_COL);
+
+ GridSqlColumn keyCol = new GridSqlColumn(h2KeyCol, tbl, h2KeyCol.getName());
+ keyCol.resultType(GridSqlType.fromColumn(h2KeyCol));
+
+ GridSqlAlias al = SplitterUtils.alias(QueryUtils.KEY_FIELD_NAME, keyCol);
+
+ return al;
+ }
+
+ /**
* @param sqlQry Query.
* @param qryAst Select AST.
* @param paramsCnt Number of parameters.
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sql/GridSqlSelect.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sql/GridSqlSelect.java
index aa91f17..85d253f 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sql/GridSqlSelect.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sql/GridSqlSelect.java
@@ -413,4 +413,28 @@ public class GridSqlSelect extends GridSqlQuery {
aliases.add((GridSqlAlias)from);
}
+
+ /**
+ * @return Copy of this select for SELECT FOR UPDATE specific tasks.
+ */
+ public GridSqlSelect copySelectForUpdate() {
+ assert isForUpdate && !distinct && havingCol < 0 && grpCols == null; // Not supported by SFU.
+
+ GridSqlSelect copy = new GridSqlSelect();
+
+ copy.from(from())
+ .where(where());
+
+ int vis = visibleColumns();
+
+ for (int i = 0; i < columns(false).size(); i++)
+ copy.addColumn(column(i), i < vis);
+
+ if (!sort().isEmpty()) {
+ for (GridSqlSortColumn sortCol : sort())
+ copy.addSort(sortCol);
+ }
+
+ return copy;
+ }
}
\ No newline at end of file
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMapQueryExecutor.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMapQueryExecutor.java
index 7314a4e..a42ab12 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMapQueryExecutor.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMapQueryExecutor.java
@@ -31,7 +31,6 @@ import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
import javax.cache.CacheException;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
@@ -46,32 +45,24 @@ import org.apache.ignite.events.Event;
import org.apache.ignite.events.EventType;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.GridTopic;
-import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.managers.communication.GridMessageListener;
import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
-import org.apache.ignite.internal.processors.cache.distributed.dht.CompoundLockFuture;
-import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTransactionalCacheAdapter;
-import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxLocalAdapter;
import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
-import org.apache.ignite.internal.processors.cache.mvcc.MvccUtils;
import org.apache.ignite.internal.processors.cache.query.CacheQueryType;
import org.apache.ignite.internal.processors.cache.query.GridCacheQueryMarshallable;
import org.apache.ignite.internal.processors.cache.query.GridCacheSqlQuery;
import org.apache.ignite.internal.processors.query.GridQueryCancel;
-import org.apache.ignite.internal.processors.query.IgniteSQLException;
import org.apache.ignite.internal.processors.query.h2.H2ConnectionWrapper;
import org.apache.ignite.internal.processors.query.h2.H2Utils;
import org.apache.ignite.internal.processors.query.h2.IgniteH2Indexing;
import org.apache.ignite.internal.processors.query.h2.MapH2QueryInfo;
-import org.apache.ignite.internal.processors.query.h2.ResultSetEnlistFuture;
import org.apache.ignite.internal.processors.query.h2.UpdateResult;
import org.apache.ignite.internal.processors.query.h2.opt.GridH2RetryException;
import org.apache.ignite.internal.processors.query.h2.opt.QueryContext;
import org.apache.ignite.internal.processors.query.h2.opt.QueryContextRegistry;
import org.apache.ignite.internal.processors.query.h2.opt.join.DistributedJoinContext;
-import org.apache.ignite.internal.processors.query.h2.sql.GridSqlQueryParser;
import org.apache.ignite.internal.processors.query.h2.twostep.messages.GridQueryCancelRequest;
import org.apache.ignite.internal.processors.query.h2.twostep.messages.GridQueryFailResponse;
import org.apache.ignite.internal.processors.query.h2.twostep.messages.GridQueryNextPageRequest;
@@ -79,17 +70,13 @@ import org.apache.ignite.internal.processors.query.h2.twostep.messages.GridQuery
import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2DmlRequest;
import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2DmlResponse;
import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2QueryRequest;
-import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2SelectForUpdateTxDetails;
import org.apache.ignite.internal.util.GridSpinBusyLock;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.U;
-import org.apache.ignite.lang.IgniteInClosure;
-import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.spi.indexing.IndexingQueryFilter;
import org.h2.api.ErrorCode;
-import org.h2.command.Prepared;
import org.h2.jdbc.JdbcResultSet;
import org.h2.jdbc.JdbcSQLException;
import org.h2.value.Value;
@@ -270,15 +257,11 @@ public class GridMapQueryExecutor {
final int[] parts = qryParts == null ? partsMap == null ? null : partsMap.get(ctx.localNodeId()) : qryParts;
- final GridDhtTxLocalAdapter tx;
-
- GridH2SelectForUpdateTxDetails txReq = req.txDetails();
-
boolean distributedJoins = req.isFlagSet(GridH2QueryRequest.FLAG_DISTRIBUTED_JOINS);
boolean enforceJoinOrder = req.isFlagSet(GridH2QueryRequest.FLAG_ENFORCE_JOIN_ORDER);
boolean explain = req.isFlagSet(GridH2QueryRequest.FLAG_EXPLAIN);
boolean replicated = req.isFlagSet(GridH2QueryRequest.FLAG_REPLICATED);
- final boolean lazy = req.isFlagSet(GridH2QueryRequest.FLAG_LAZY) && txReq == null;
+ final boolean lazy = req.isFlagSet(GridH2QueryRequest.FLAG_LAZY);
Boolean dataPageScanEnabled = req.isDataPageScanEnabled();
@@ -289,66 +272,6 @@ public class GridMapQueryExecutor {
final Object[] params = req.parameters();
- try {
- if (txReq != null) {
- // Prepare to run queries.
- GridCacheContext<?, ?> mainCctx = mainCacheContext(cacheIds);
-
- if (mainCctx == null || mainCctx.atomic() || !mainCctx.mvccEnabled() || cacheIds.size() != 1)
- throw new IgniteSQLException("SELECT FOR UPDATE is supported only for queries " +
- "that involve single transactional cache.");
-
- GridDhtTransactionalCacheAdapter txCache = (GridDhtTransactionalCacheAdapter)mainCctx.cache();
-
- if (!node.isLocal()) {
- tx = txCache.initTxTopologyVersion(
- node.id(),
- node,
- txReq.version(),
- txReq.futureId(),
- txReq.miniId(),
- txReq.firstClientRequest(),
- req.topologyVersion(),
- txReq.threadId(),
- txReq.timeout(),
- txReq.subjectId(),
- txReq.taskNameHash(),
- req.mvccSnapshot());
- }
- else {
- tx = MvccUtils.tx(ctx, txReq.version());
-
- assert tx != null;
- }
- }
- else
- tx = null;
- }
- catch (IgniteException | IgniteCheckedException e) {
- // send error if TX was not initialized properly.
- releaseReservations();
-
- U.error(log, "Failed to execute local query.", e);
-
- sendError(node, req.requestId(), e);
-
- return;
- }
-
- AtomicInteger runCntr;
- CompoundLockFuture lockFut;
-
- if (txReq != null && segments > 1) {
- runCntr = new AtomicInteger(segments);
- lockFut = new CompoundLockFuture(segments, tx);
-
- lockFut.init();
- }
- else {
- runCntr = null;
- lockFut = null;
- }
-
for (int i = 1; i < segments; i++) {
assert !F.isEmpty(cacheIds);
@@ -374,10 +297,6 @@ public class GridMapQueryExecutor {
params,
lazy,
req.mvccSnapshot(),
- tx,
- txReq,
- lockFut,
- runCntr,
dataPageScanEnabled);
return null;
@@ -403,10 +322,6 @@ public class GridMapQueryExecutor {
params,
lazy,
req.mvccSnapshot(),
- tx,
- txReq,
- lockFut,
- runCntr,
dataPageScanEnabled);
}
@@ -428,10 +343,6 @@ public class GridMapQueryExecutor {
* @param params Query parameters.
* @param lazy Streaming flag.
* @param mvccSnapshot MVCC snapshot.
- * @param tx Transaction.
- * @param txDetails TX details, if it's a {@code FOR UPDATE} request, or {@code null}.
- * @param lockFut Lock future.
- * @param runCntr Counter which counts remaining queries in case segmented index is used.
* @param dataPageScanEnabled If data page scan is enabled.
*/
private void onQueryRequest0(
@@ -452,18 +363,8 @@ public class GridMapQueryExecutor {
final Object[] params,
boolean lazy,
@Nullable final MvccSnapshot mvccSnapshot,
- @Nullable final GridDhtTxLocalAdapter tx,
- @Nullable final GridH2SelectForUpdateTxDetails txDetails,
- @Nullable final CompoundLockFuture lockFut,
- @Nullable final AtomicInteger runCntr,
- Boolean dataPageScanEnabled) {
- // In presence of TX, we also must always have matching details.
- assert tx == null || txDetails != null;
-
- assert !lazy || txDetails == null : "Lazy execution of SELECT FOR UPDATE queries is not supported.";
-
- boolean inTx = (tx != null);
-
+ Boolean dataPageScanEnabled
+ ) {
// Prepare to run queries.
GridCacheContext<?, ?> mainCctx = mainCacheContext(cacheIds);
@@ -474,9 +375,7 @@ public class GridMapQueryExecutor {
PartitionReservation reserved = null;
try {
- // We want to reserve only in not SELECT FOR UPDATE case -
- // otherwise, their state is protected by locked topology.
- if (topVer != null && txDetails == null) {
+ if (topVer != null) {
// Reserve primary for topology version or explicit partitions.
reserved = h2.partitionReservationManager().reservePartitions(
cacheIds,
@@ -515,7 +414,7 @@ public class GridMapQueryExecutor {
reserved
);
- qryResults = new MapQueryResults(h2, reqId, qrys.size(), mainCctx, inTx, lazy, qctx);
+ qryResults = new MapQueryResults(h2, reqId, qrys.size(), mainCctx, lazy, qctx);
// qctx is set, we have to release reservations inside of it.
reserved = null;
@@ -555,12 +454,11 @@ public class GridMapQueryExecutor {
qryResults.addResult(qryIdx, res);
+ ResultSet rs = null;
+
try {
res.lock();
- boolean removeMapping = false;
- ResultSet rs = null;
-
MapH2QueryInfo qryInfo = null;
// If we are not the target node for this replicated query, just ignore it.
@@ -577,17 +475,8 @@ public class GridMapQueryExecutor {
throw new IgniteCheckedException("Failed to parse SQL query: " + sql, e);
}
- Prepared p = GridSqlQueryParser.prepared(stmt);
-
- if (GridSqlQueryParser.isForUpdateQuery(p)) {
- sql = GridSqlQueryParser.rewriteQueryForUpdateIfNeeded(p, inTx);
- stmt = h2.connections().prepareStatement(connWrp.connection(), sql);
- }
-
H2Utils.bindParameters(stmt, params0);
- int opTimeout = IgniteH2Indexing.operationTimeout(timeout, tx);
-
qryInfo = new MapH2QueryInfo(stmt, qry.query(), node, reqId, segmentId);
rs = h2.executeSqlQueryWithTimer(
@@ -595,36 +484,11 @@ public class GridMapQueryExecutor {
connWrp.connection(),
sql,
params0,
- opTimeout,
+ timeout,
qryResults.queryCancel(qryIdx),
dataPageScanEnabled,
qryInfo);
- if (inTx) {
- ResultSetEnlistFuture enlistFut = ResultSetEnlistFuture.future(
- ctx.localNodeId(),
- txDetails.version(),
- mvccSnapshot,
- txDetails.threadId(),
- IgniteUuid.randomUuid(),
- txDetails.miniId(),
- parts,
- tx,
- opTimeout,
- mainCctx,
- rs
- );
-
- if (lockFut != null)
- lockFut.register(enlistFut);
-
- enlistFut.init();
-
- enlistFut.get();
-
- rs.beforeFirst();
- }
-
if (evt) {
ctx.event().record(new CacheQueryExecutedEvent<>(
node,
@@ -649,13 +513,6 @@ public class GridMapQueryExecutor {
if (qryResults.cancelled())
throw new QueryCancelledException();
- if (inTx) {
- if (tx.dht() && (runCntr == null || runCntr.decrementAndGet() == 0)) {
- if (removeMapping = tx.empty() && !tx.queryEnlisted())
- ctx.cache().context().tm().forgetTx(tx);
- }
- }
-
final GridQueryNextPageResponse msg = prepareNextPage(
nodeRess,
node,
@@ -663,30 +520,10 @@ public class GridMapQueryExecutor {
qryIdx,
segmentId,
pageSize,
- removeMapping,
dataPageScanEnabled
);
- // Send the first page.
- if (lockFut == null)
- sendNextPage(node, msg);
- else {
- if (msg != null) {
- lockFut.listen(new IgniteInClosure<IgniteInternalFuture<Void>>() {
- @Override public void apply(IgniteInternalFuture<Void> future) {
- try {
- if (node.isLocal())
- h2.reduceQueryExecutor().onMessage(ctx.localNodeId(), msg);
- else
- ctx.io().sendToGridTopic(node, GridTopic.TOPIC_QUERY, msg, QUERY_POOL);
- }
- catch (Exception e) {
- U.error(log, e);
- }
- }
- });
- }
- }
+ sendNextPage(node, msg);
qryIdx++;
}
@@ -967,7 +804,7 @@ public class GridMapQueryExecutor {
sendError(node, reqId, new CacheException("No query result found for request: " + req));
else if (qryResults.cancelled())
sendQueryCancel(node, reqId);
- else
+ else {
try {
QueryContext qctxReduce = qryCtxRegistry.getThreadLocal();
@@ -996,7 +833,6 @@ public class GridMapQueryExecutor {
req.query(),
req.segmentId(),
req.pageSize(),
- false,
dataPageScanEnabled);
sendNextPage(node, msg);
@@ -1031,6 +867,7 @@ public class GridMapQueryExecutor {
qryResults.cancel();
}
+ }
}
/**
@@ -1040,7 +877,6 @@ public class GridMapQueryExecutor {
* @param qry Query.
* @param segmentId Index segment ID.
* @param pageSize Page size.
- * @param removeMapping Remove mapping flag.
* @param dataPageScanEnabled If data page scan is enabled.
* @return Next page.
* @throws IgniteCheckedException If failed.
@@ -1052,7 +888,6 @@ public class GridMapQueryExecutor {
int qry,
int segmentId,
int pageSize,
- boolean removeMapping,
Boolean dataPageScanEnabled) throws IgniteCheckedException {
MapQueryResult res = qr.result(qry);
@@ -1081,19 +916,13 @@ public class GridMapQueryExecutor {
boolean loc = node.isLocal();
- // In case of SELECT FOR UPDATE the last columns is _KEY,
- // we can't retrieve them for an arbitrary row otherwise.
- int colsCnt = !qr.isForUpdate() ? res.columnCount() : res.columnCount() - 1;
-
GridQueryNextPageResponse msg = new GridQueryNextPageResponse(qr.queryRequestId(), segmentId, qry, page,
page == 0 ? res.rowCount() : -1,
- colsCnt,
- loc ? null : toMessages(rows, new ArrayList<>(res.columnCount()), colsCnt),
+ res.columnCount(),
+ loc ? null : toMessages(rows, new ArrayList<>(res.columnCount()), res.columnCount()),
loc ? rows : null,
last);
- msg.removeMapping(removeMapping);
-
return msg;
}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java
index 5aab4b5..1fe7ed4 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java
@@ -33,7 +33,6 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@@ -57,16 +56,12 @@ import org.apache.ignite.internal.managers.communication.GridMessageListener;
import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
-import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
-import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxSelectForUpdateFuture;
-import org.apache.ignite.internal.processors.cache.distributed.near.TxTopologyVersionFuture;
import org.apache.ignite.internal.processors.cache.mvcc.MvccQueryTracker;
import org.apache.ignite.internal.processors.cache.query.GridCacheQueryMarshallable;
import org.apache.ignite.internal.processors.cache.query.GridCacheSqlQuery;
import org.apache.ignite.internal.processors.cache.query.GridCacheTwoStepQuery;
import org.apache.ignite.internal.processors.query.GridQueryCacheObjectsIterator;
import org.apache.ignite.internal.processors.query.GridQueryCancel;
-import org.apache.ignite.internal.processors.query.IgniteSQLException;
import org.apache.ignite.internal.processors.query.h2.H2ConnectionWrapper;
import org.apache.ignite.internal.processors.query.h2.H2FieldsIterator;
import org.apache.ignite.internal.processors.query.h2.H2Utils;
@@ -86,7 +81,6 @@ import org.apache.ignite.internal.processors.query.h2.twostep.messages.GridQuery
import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2DmlRequest;
import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2DmlResponse;
import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2QueryRequest;
-import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2SelectForUpdateTxDetails;
import org.apache.ignite.internal.transactions.IgniteTxAlreadyCompletedCheckedException;
import org.apache.ignite.internal.util.GridSpinBusyLock;
import org.apache.ignite.internal.util.typedef.C2;
@@ -96,7 +90,6 @@ import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteBiClosure;
import org.apache.ignite.lang.IgniteFuture;
-import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.transactions.TransactionAlreadyCompletedException;
import org.apache.ignite.transactions.TransactionException;
@@ -112,7 +105,6 @@ import org.jetbrains.annotations.Nullable;
import static java.util.Collections.singletonList;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_SQL_RETRY_TIMEOUT;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.checkActive;
-import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.mvccEnabled;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.tx;
import static org.apache.ignite.internal.processors.cache.query.GridCacheSqlQuery.EMPTY_PARAMS;
import static org.apache.ignite.internal.processors.query.h2.sql.GridSqlQuerySplitter.mergeTableIdentifier;
@@ -370,15 +362,8 @@ public class GridReduceQueryExecutor {
if (msg.retry() != null)
r.setStateOnRetry(node.id(), msg.retry(), msg.retryCause());
- else if (msg.page() == 0) {
- // Do count down on each first page received.
+ else if (msg.page() == 0) // Count down only on each first page received.
r.latch().countDown();
-
- GridNearTxSelectForUpdateFuture sfuFut = r.selectForUpdateFuture();
-
- if (sfuFut != null)
- sfuFut.onResult(node.id(), (long)msg.allRows(), msg.removeMapping(), null);
- }
}
/**
@@ -407,7 +392,6 @@ public class GridReduceQueryExecutor {
* @param lazy Lazy execution flag.
* @param mvccTracker Query tracker.
* @param dataPageScanEnabled If data page scan is enabled.
- * @param forUpdate For update flag.
* @param pageSize Page size.
* @return Rows iterator.
*/
@@ -424,7 +408,6 @@ public class GridReduceQueryExecutor {
boolean lazy,
MvccQueryTracker mvccTracker,
Boolean dataPageScanEnabled,
- boolean forUpdate,
int pageSize
) {
// If explicit partitions are set, but there are no real tables, ignore.
@@ -433,6 +416,14 @@ public class GridReduceQueryExecutor {
assert !qry.mvccEnabled() || mvccTracker != null;
+ try {
+ if (qry.mvccEnabled())
+ checkActive(tx(ctx));
+ }
+ catch (IgniteTxAlreadyCompletedCheckedException e) {
+ throw new TransactionAlreadyCompletedException(e.getMessage(), e);
+ }
+
final boolean singlePartMode = parts != null && parts.length == 1;
if (F.isEmpty(params))
@@ -482,52 +473,12 @@ public class GridReduceQueryExecutor {
List<Integer> cacheIds = qry.cacheIds();
- boolean mvccEnabled = mvccEnabled(ctx);
+ AffinityTopologyVersion topVer = h2.readyTopologyVersion();
- final GridNearTxLocal curTx;
-
- try {
- curTx = mvccEnabled ? checkActive(tx(ctx)) : null;
- }
- catch (IgniteTxAlreadyCompletedCheckedException e) {
- throw new TransactionAlreadyCompletedException(e.getMessage(), e);
- }
-
- final GridNearTxSelectForUpdateFuture sfuFut;
-
- final boolean clientFirst;
-
- AffinityTopologyVersion topVer;
-
- if (forUpdate) {
- // Indexing should have started TX at this point for FOR UPDATE query.
- assert mvccEnabled && curTx != null;
-
- try {
- TxTopologyVersionFuture topFut = new TxTopologyVersionFuture(curTx, mvccTracker.context());
-
- topVer = topFut.get();
-
- clientFirst = topFut.clientFirst();
- }
- catch (IgniteCheckedException e) {
- throw new IgniteSQLException("Failed to map SELECT FOR UPDATE query on topology.", e);
- }
-
- sfuFut = new GridNearTxSelectForUpdateFuture(mvccTracker.context(), curTx, timeoutMillis);
- }
- else {
- sfuFut = null;
-
- clientFirst = false;
-
- topVer = h2.readyTopologyVersion();
-
- // Check if topology has changed while retrying on locked topology.
- if (h2.serverTopologyChanged(topVer) && ctx.cache().context().lockedTopologyVersion(null) != null) {
- throw new CacheException(new TransactionException("Server topology is changed during query " +
- "execution inside a transaction. It's recommended to rollback and retry transaction."));
- }
+ // Check if topology has changed while retrying on locked topology.
+ if (h2.serverTopologyChanged(topVer) && ctx.cache().context().lockedTopologyVersion(null) != null) {
+ throw new CacheException(new TransactionException("Server topology is changed during query " +
+ "execution inside a transaction. It's recommended to rollback and retry transaction."));
}
long qryReqId = qryIdGen.incrementAndGet();
@@ -536,7 +487,6 @@ public class GridReduceQueryExecutor {
h2.connections().connectionForThread().connection(schemaName),
mapQueries.size(),
pageSize,
- sfuFut,
dataPageScanEnabled
);
@@ -576,12 +526,8 @@ public class GridReduceQueryExecutor {
partsMap = nodesParts.partitionsMap();
qryMap = nodesParts.queryPartitionsMap();
- if (nodes == null) {
- if (sfuFut != null)
- sfuFut.onDone(0L, null);
-
+ if (nodes == null)
continue; // Retry.
- }
assert !nodes.isEmpty();
@@ -599,9 +545,6 @@ public class GridReduceQueryExecutor {
}
}
- if (sfuFut != null && !sfuFut.isFailed())
- sfuFut.init(topVer, nodes);
-
int tblIdx = 0;
final boolean skipMergeTbl = !qry.explain() && qry.skipMergeTable() || singlePartMode;
@@ -717,44 +660,11 @@ public class GridReduceQueryExecutor {
.timeout(timeoutMillis)
.schemaName(schemaName);
- if (curTx != null && curTx.mvccSnapshot() != null)
- req.mvccSnapshot(curTx.mvccSnapshot());
- else if (mvccTracker != null)
+ if (mvccTracker != null)
req.mvccSnapshot(mvccTracker.snapshot());
- final C2<ClusterNode, Message, Message> pspec =
- (parts == null ? null : new ReducePartitionsSpecializer(qryMap));
-
- final C2<ClusterNode, Message, Message> spec;
-
- if (forUpdate) {
- final AtomicInteger cnt = new AtomicInteger();
-
- spec = new C2<ClusterNode, Message, Message>() {
- @Override public Message apply(ClusterNode clusterNode, Message msg) {
- assert msg instanceof GridH2QueryRequest;
-
- GridH2QueryRequest res = pspec != null ? (GridH2QueryRequest)pspec.apply(clusterNode, msg) :
- new GridH2QueryRequest((GridH2QueryRequest)msg);
-
- GridH2SelectForUpdateTxDetails txReq = new GridH2SelectForUpdateTxDetails(
- curTx.threadId(),
- IgniteUuid.randomUuid(),
- cnt.incrementAndGet(),
- curTx.subjectId(),
- curTx.xidVersion(),
- curTx.taskNameHash(),
- clientFirst,
- curTx.remainingTime());
-
- res.txDetails(txReq);
-
- return res;
- }
- };
- }
- else
- spec = pspec;
+ final C2<ClusterNode, Message, Message> spec =
+ parts == null ? null : new ReducePartitionsSpecializer(qryMap);
if (send(nodes, req, spec, false)) {
awaitAllReplies(r, nodes, cancel);
@@ -774,9 +684,6 @@ public class GridReduceQueryExecutor {
else {
retry = true;
- // On-the-fly topology change must not be possible in FOR UPDATE case.
- assert sfuFut == null;
-
// If remote node asks us to retry then we have outdated full partition map.
h2.awaitForReadyTopologyVersion(r.retryTopologyVersion());
}
@@ -837,7 +744,7 @@ public class GridReduceQueryExecutor {
qryInfo
);
- resIter = new H2FieldsIterator(res, mvccTracker, false, detachedConn);
+ resIter = new H2FieldsIterator(res, mvccTracker, detachedConn);
// don't recycle at final block
detachedConn = null;
@@ -856,15 +763,9 @@ public class GridReduceQueryExecutor {
if (Thread.currentThread().isInterrupted())
throw new IgniteInterruptedCheckedException("Query was interrupted.");
- if (sfuFut != null)
- sfuFut.onDone(0L);
-
continue;
}
- if (sfuFut != null)
- sfuFut.get();
-
return new GridQueryCacheObjectsIterator(resIter, h2.objectContext(), keepBinary);
}
catch (IgniteCheckedException | RuntimeException e) {
@@ -872,21 +773,12 @@ public class GridReduceQueryExecutor {
U.closeQuiet(r.connection());
- CacheException resEx = null;
-
if (e instanceof CacheException) {
if (wasCancelled((CacheException)e))
- resEx = new CacheException("Failed to run reduce query locally.",
+ throw new CacheException("Failed to run reduce query locally.",
new QueryCancelledException());
- else
- resEx = (CacheException)e;
- }
- if (resEx != null) {
- if (sfuFut != null)
- sfuFut.onDone(resEx);
-
- throw resEx;
+ throw (CacheException)e;
}
Throwable cause = e;
@@ -899,12 +791,7 @@ public class GridReduceQueryExecutor {
cause = disconnectedErr;
}
- resEx = new CacheException("Failed to run reduce query locally.", cause);
-
- if (sfuFut != null)
- sfuFut.onDone(resEx);
-
- throw resEx;
+ throw new CacheException("Failed to run reduce query locally.", cause);
}
finally {
if (detachedConn != null)
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/MapQueryResults.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/MapQueryResults.java
index d61e09a..f42ea61 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/MapQueryResults.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/MapQueryResults.java
@@ -18,7 +18,6 @@
package org.apache.ignite.internal.processors.query.h2.twostep;
import java.util.concurrent.atomic.AtomicReferenceArray;
-import org.apache.ignite.IgniteLogger;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.query.GridQueryCancel;
import org.apache.ignite.internal.processors.query.h2.IgniteH2Indexing;
@@ -50,9 +49,6 @@ class MapQueryResults {
/** */
private volatile boolean cancelled;
- /** {@code SELECT FOR UPDATE} flag. */
- private final boolean forUpdate;
-
/** Query context. */
private final QueryContext qctx;
@@ -63,13 +59,11 @@ class MapQueryResults {
* @param qryReqId Query request ID.
* @param qrys Number of queries.
* @param cctx Cache context.
- * @param forUpdate {@code SELECT FOR UPDATE} flag.
* @param lazy Lazy flag.
* @param qctx Query context.
*/
MapQueryResults(IgniteH2Indexing h2, long qryReqId, int qrys, @Nullable GridCacheContext<?, ?> cctx,
- boolean forUpdate, boolean lazy, QueryContext qctx) {
- this.forUpdate = forUpdate;
+ boolean lazy, QueryContext qctx) {
this.h2 = h2;
this.qryReqId = qryReqId;
this.cctx = cctx;
@@ -205,13 +199,6 @@ class MapQueryResults {
}
/**
- * @return {@code SELECT FOR UPDATE} flag.
- */
- public boolean isForUpdate() {
- return forUpdate;
- }
-
- /**
* @return Query context.
*/
public QueryContext queryContext() {
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/ReduceQueryRun.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/ReduceQueryRun.java
index ccd14c3..1fb3c78 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/ReduceQueryRun.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/ReduceQueryRun.java
@@ -27,7 +27,6 @@ import javax.cache.CacheException;
import org.apache.ignite.cache.query.Query;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
-import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxSelectForUpdateFuture;
import org.apache.ignite.internal.util.typedef.F;
import org.h2.jdbc.JdbcConnection;
import org.jetbrains.annotations.Nullable;
@@ -54,22 +53,17 @@ public class ReduceQueryRun {
/** */
private final AtomicReference<State> state = new AtomicReference<>();
- /** Future controlling {@code SELECT FOR UPDATE} query execution. */
- private final GridNearTxSelectForUpdateFuture selectForUpdateFut;
-
/**
* Constructor.
* @param conn Connection.
* @param idxsCnt Number of indexes.
* @param pageSize Page size.
- * @param selectForUpdateFut Future controlling {@code SELECT FOR UPDATE} query execution.
* @param dataPageScanEnabled If data page scan is enabled.
*/
ReduceQueryRun(
Connection conn,
int idxsCnt,
int pageSize,
- GridNearTxSelectForUpdateFuture selectForUpdateFut,
Boolean dataPageScanEnabled
) {
this.conn = (JdbcConnection)conn;
@@ -77,7 +71,6 @@ public class ReduceQueryRun {
idxs = new ArrayList<>(idxsCnt);
this.pageSize = pageSize > 0 ? pageSize : Query.DFLT_PAGE_SIZE;
- this.selectForUpdateFut = selectForUpdateFut;
this.dataPageScanEnabled = dataPageScanEnabled;
}
@@ -221,13 +214,6 @@ public class ReduceQueryRun {
}
/**
- * @return {@code SELECT FOR UPDATE} future, if any.
- */
- @Nullable public GridNearTxSelectForUpdateFuture selectForUpdateFuture() {
- return selectForUpdateFut;
- }
-
- /**
* Error state.
*/
private static class State {
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/ReduceResultPage.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/ReduceResultPage.java
index 1a25c50..9c82253 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/ReduceResultPage.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/ReduceResultPage.java
@@ -17,11 +17,9 @@
package org.apache.ignite.internal.processors.query.h2.twostep;
-import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
-import java.util.List;
import java.util.NoSuchElementException;
import java.util.RandomAccess;
import java.util.UUID;
@@ -29,9 +27,7 @@ import javax.cache.CacheException;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.processors.query.h2.twostep.messages.GridQueryNextPageResponse;
-import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.S;
-import org.apache.ignite.lang.IgniteClosure;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.h2.value.Value;
@@ -77,21 +73,7 @@ public class ReduceResultPage {
rowsInPage = plainRows.size();
- if (rowsInPage == 0 || ((List<Value[]>)plainRows).get(0).length == res.columns())
- rows = (Iterator<Value[]>)plainRows.iterator();
- else {
- // If it's a result of SELECT FOR UPDATE (we can tell by difference in number
- // of columns checked above), we need to strip off stuff we don't need.
- rows = F.iterator(plainRows, new IgniteClosure<Object, Value[]>() {
- @Override public Value[] apply(Object o) {
- Value[] row = (Value[])o;
-
- assert row.length >= res.columns();
-
- return Arrays.copyOfRange(row, 0, res.columns());
- }
- }, true);
- }
+ rows = (Iterator<Value[]>)plainRows.iterator();
}
else {
final int cols = res.columns();
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccPartitionedSelectForUpdateQueryTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccPartitionedSelectForUpdateQueryTest.java
deleted file mode 100644
index 139359f..0000000
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccPartitionedSelectForUpdateQueryTest.java
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
- * 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.cache.mvcc;
-
-import org.apache.ignite.cache.CacheMode;
-import org.junit.Test;
-
-import static org.apache.ignite.cache.CacheMode.PARTITIONED;
-
-/** */
-public class CacheMvccPartitionedSelectForUpdateQueryTest extends CacheMvccSelectForUpdateQueryAbstractTest {
- /** {@inheritDoc} */
- @Override public CacheMode cacheMode() {
- return PARTITIONED;
- }
-
- /**
- *
- */
- @Test
- public void testSelectForUpdateDistributedSegmented() throws Exception {
- doTestSelectForUpdateDistributed("PersonSeg", false);
- }
-
- /**
- *
- */
- @Test
- public void testSelectForUpdateLocalSegmented() throws Exception {
- doTestSelectForUpdateLocal("PersonSeg", false);
- }
-}
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccReplicatedSelectForUpdateQueryTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccReplicatedSelectForUpdateQueryTest.java
deleted file mode 100644
index b123e57..0000000
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccReplicatedSelectForUpdateQueryTest.java
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * 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.cache.mvcc;
-
-import org.apache.ignite.cache.CacheMode;
-
-import static org.apache.ignite.cache.CacheMode.REPLICATED;
-
-/** */
-public class CacheMvccReplicatedSelectForUpdateQueryTest extends CacheMvccSelectForUpdateQueryAbstractTest {
- /** {@inheritDoc} */
- @Override public CacheMode cacheMode() {
- return REPLICATED;
- }
-}
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccSelectForUpdateQueryAbstractTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccSelectForUpdateQueryAbstractTest.java
deleted file mode 100644
index 7bec577..0000000
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccSelectForUpdateQueryAbstractTest.java
+++ /dev/null
@@ -1,370 +0,0 @@
-/*
- * 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.cache.mvcc;
-
-import java.sql.Connection;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
-import java.util.concurrent.Callable;
-import java.util.concurrent.TimeUnit;
-import javax.cache.CacheException;
-import org.apache.ignite.Ignite;
-import org.apache.ignite.IgniteCache;
-import org.apache.ignite.cache.QueryEntity;
-import org.apache.ignite.cache.query.FieldsQueryCursor;
-import org.apache.ignite.cache.query.SqlFieldsQuery;
-import org.apache.ignite.configuration.CacheConfiguration;
-import org.apache.ignite.internal.IgniteEx;
-import org.apache.ignite.internal.IgniteInternalFuture;
-import org.apache.ignite.internal.processors.query.IgniteSQLException;
-import org.apache.ignite.internal.util.typedef.X;
-import org.apache.ignite.internal.util.typedef.internal.U;
-import org.apache.ignite.testframework.GridTestUtils;
-import org.apache.ignite.transactions.Transaction;
-import org.apache.ignite.transactions.TransactionConcurrency;
-import org.apache.ignite.transactions.TransactionIsolation;
-import org.junit.Test;
-
-import static org.apache.ignite.cache.CacheMode.PARTITIONED;
-import static org.apache.ignite.internal.processors.cache.index.AbstractSchemaSelfTest.connect;
-import static org.apache.ignite.internal.processors.cache.index.AbstractSchemaSelfTest.execute;
-
-/**
- * Test for {@code SELECT FOR UPDATE} queries.
- */
-public abstract class CacheMvccSelectForUpdateQueryAbstractTest extends CacheMvccAbstractTest {
- /** */
- private static final int CACHE_SIZE = 50;
-
- /** {@inheritDoc} */
- @SuppressWarnings("unchecked")
- @Override protected void beforeTest() throws Exception {
- super.beforeTest();
-
- disableScheduledVacuum = getName().equals("testSelectForUpdateAfterAbortedTx");
-
- IgniteEx grid = startGrid(0);
-
- CacheConfiguration seg = new CacheConfiguration("segmented*");
-
- seg.setCacheMode(cacheMode());
-
- if (seg.getCacheMode() == PARTITIONED)
- seg.setQueryParallelism(4);
-
- grid.addCacheConfiguration(seg);
-
- try (Connection c = connect(grid)) {
- execute(c, "create table person (id int primary key, firstName varchar, lastName varchar) " +
- "with \"atomicity=transactional_snapshot,cache_name=Person\"");
-
- execute(c, "create table person_seg (id int primary key, firstName varchar, lastName varchar) " +
- "with \"atomicity=transactional_snapshot,cache_name=PersonSeg,template=segmented\"");
-
- try (Transaction tx = grid(0).transactions().txStart(TransactionConcurrency.PESSIMISTIC,
- TransactionIsolation.REPEATABLE_READ)) {
-
- for (int i = 1; i <= CACHE_SIZE; i++) {
- execute(c, "insert into person(id, firstName, lastName) values(" + i + ",'" + i + "','" + i + "')");
-
- execute(c, "insert into person_seg(id, firstName, lastName) " +
- "values(" + i + ",'" + i + "','" + i + "')");
- }
-
- tx.commit();
- }
- }
-
- startGridsMultiThreaded(1, 2);
- }
-
- /**
- *
- */
- @Test
- public void testSelectForUpdateDistributed() throws Exception {
- doTestSelectForUpdateDistributed("Person", false);
- }
-
-
- /**
- * @throws Exception If failed.
- */
- @Test
- public void testSelectForUpdateLocal() throws Exception {
- doTestSelectForUpdateLocal("Person", false);
- }
-
- /**
- * @throws Exception If failed.
- */
- @Test
- public void testSelectForUpdateOutsideTxDistributed() throws Exception {
- doTestSelectForUpdateDistributed("Person", true);
- }
-
- /**
- * @throws Exception If failed.
- */
- @Test
- public void testSelectForUpdateOutsideTxLocal() throws Exception {
- doTestSelectForUpdateLocal("Person", true);
- }
-
- /**
- * @param cacheName Cache name.
- * @param outsideTx Whether select is executed outside transaction
- * @throws Exception If failed.
- */
- void doTestSelectForUpdateLocal(String cacheName, boolean outsideTx) throws Exception {
- Ignite node = grid(0);
-
- IgniteCache<Integer, ?> cache = node.cache(cacheName);
-
- Transaction ignored = outsideTx ? null : node.transactions().txStart(TransactionConcurrency.PESSIMISTIC,
- TransactionIsolation.REPEATABLE_READ);
-
- try {
- SqlFieldsQuery qry = new SqlFieldsQuery("select id, * from " + tableName(cache) + " order by id for update")
- .setLocal(true);
-
- FieldsQueryCursor<List<?>> query = cache.query(qry);
-
- List<List<?>> res = query.getAll();
-
- List<Integer> keys = new ArrayList<>();
-
- for (List<?> r : res)
- keys.add((Integer)r.get(0));
-
- checkLocks(cacheName, keys, !outsideTx);
- }
- finally {
- U.close(ignored, log);
- }
- }
-
- /**
- * @param cacheName Cache name.
- * @param outsideTx Whether select is executed outside transaction
- * @throws Exception If failed.
- */
- void doTestSelectForUpdateDistributed(String cacheName, boolean outsideTx) throws Exception {
- awaitPartitionMapExchange();
-
- Ignite node = grid(0);
-
- IgniteCache<Integer, ?> cache = node.cache(cacheName);
-
- Transaction ignored = outsideTx ? null : node.transactions().txStart(TransactionConcurrency.PESSIMISTIC,
- TransactionIsolation.REPEATABLE_READ);
-
- try {
- SqlFieldsQuery qry = new SqlFieldsQuery("select id, * from " + tableName(cache) + " order by id for update")
- .setPageSize(10);
-
- FieldsQueryCursor<List<?>> query = cache.query(qry);
-
- List<List<?>> res = query.getAll();
-
- List<Integer> keys = new ArrayList<>();
-
- for (List<?> r : res)
- keys.add((Integer)r.get(0));
-
- checkLocks(cacheName, keys, !outsideTx);
- }
- finally {
- U.close(ignored, log);
- }
- }
-
- /**
- *
- */
- @Test
- public void testSelectForUpdateWithUnion() {
- assertQueryThrows("select id from person union select 1 for update",
- "SELECT UNION FOR UPDATE is not supported.");
- }
-
- /**
- *
- */
- @Test
- public void testSelectForUpdateWithJoin() {
- assertQueryThrows("select p1.id from person p1 join person p2 on p1.id = p2.id for update",
- "SELECT FOR UPDATE with joins is not supported.");
- }
-
- /**
- *
- */
- @Test
- public void testSelectForUpdateWithLimit() {
- assertQueryThrows("select id from person limit 0,5 for update",
- "LIMIT/OFFSET clauses are not supported for SELECT FOR UPDATE.");
- }
-
- /**
- *
- */
- @Test
- public void testSelectForUpdateWithGroupings() {
- assertQueryThrows("select count(*) from person for update",
- "SELECT FOR UPDATE with aggregates and/or GROUP BY is not supported.");
-
- assertQueryThrows("select lastName, count(*) from person group by lastName for update",
- "SELECT FOR UPDATE with aggregates and/or GROUP BY is not supported.");
- }
-
- /**
- * @throws Exception If failed.
- */
- @Test
- public void testSelectForUpdateAfterAbortedTx() throws Exception {
- assert disableScheduledVacuum;
-
- Ignite node = grid(0);
-
- IgniteCache<Integer, ?> cache = node.cache("Person");
-
- List<List<?>> res;
-
- try (Transaction tx = node.transactions().txStart(TransactionConcurrency.PESSIMISTIC,
- TransactionIsolation.REPEATABLE_READ)) {
-
- res = cache.query(new SqlFieldsQuery("update person set lastName=UPPER(lastName)")).getAll();
-
- assertEquals((long)CACHE_SIZE, res.get(0).get(0));
-
- tx.rollback();
- }
-
- try (Transaction tx = node.transactions().txStart(TransactionConcurrency.PESSIMISTIC,
- TransactionIsolation.REPEATABLE_READ)) {
-
- res = cache.query(new SqlFieldsQuery("select id, * from person order by id for update")).getAll();
-
- assertEquals(CACHE_SIZE, res.size());
-
- List<Integer> keys = new ArrayList<>();
-
- for (List<?> r : res)
- keys.add((Integer)r.get(0));
-
- checkLocks("Person", keys, true);
-
- tx.rollback();
-
- checkLocks("Person", keys, false);
- }
- }
-
- /**
- * Check that an attempt to get a lock on any key from given list fails by timeout.
- *
- * @param cacheName Cache name to check.
- * @param keys Keys to check.
- * @param locked Whether the key is locked
- * @throws Exception if failed.
- */
- @SuppressWarnings({"unchecked"})
- private void checkLocks(String cacheName, List<Integer> keys, boolean locked) throws Exception {
- Ignite node = ignite(2);
- IgniteCache cache = node.cache(cacheName);
-
- List<IgniteInternalFuture<Integer>> calls = new ArrayList<>();
-
- for (int key : keys) {
- calls.add(GridTestUtils.runAsync(new Callable<Integer>() {
- /** {@inheritDoc} */
- @Override public Integer call() {
- try (Transaction ignored = node.transactions().txStart(TransactionConcurrency.PESSIMISTIC,
- TransactionIsolation.REPEATABLE_READ)) {
- List<List<?>> res = cache
- .query(
- new SqlFieldsQuery("select * from " + tableName(cache) +
- " where id = " + key + " for update").setTimeout(1, TimeUnit.SECONDS)
- )
- .getAll();
-
- return (Integer)res.get(0).get(0);
- }
- }
- }));
- }
-
- for (IgniteInternalFuture fut : calls) {
- if (!locked)
- fut.get(TX_TIMEOUT);
- else {
- try {
- fut.get();
- }
- catch (Exception e) {
- CacheException e0 = X.cause(e, CacheException.class);
-
- assert e0 != null;
-
- assert e0.getMessage() != null &&
- e0.getMessage().contains("Failed to acquire lock within provided timeout");
- }
- }
- }
- }
-
- /**
- * @param cache Cache.
- * @return Name of the table contained by this cache.
- */
- @SuppressWarnings("unchecked")
- private static String tableName(IgniteCache<?, ?> cache) {
- return ((Collection<QueryEntity>)cache.getConfiguration(CacheConfiguration.class).getQueryEntities())
- .iterator().next().getTableName();
- }
-
- /**
- * Test that query throws exception with expected message.
- * @param qry SQL.
- * @param exMsg Expected message.
- */
- private void assertQueryThrows(String qry, String exMsg) {
- assertQueryThrows(qry, exMsg, false);
-
- assertQueryThrows(qry, exMsg, true);
- }
-
- /**
- * Test that query throws exception with expected message.
- * @param qry SQL.
- * @param exMsg Expected message.
- * @param loc Local query flag.
- */
- @SuppressWarnings("ThrowableNotThrown")
- private void assertQueryThrows(String qry, String exMsg, boolean loc) {
- Ignite node = grid(0);
-
- GridTestUtils.assertThrows(null, new Callable<Object>() {
- @Override public Object call() {
- return node.cache("Person").query(new SqlFieldsQuery(qry).setLocal(loc)).getAll();
- }
- }, IgniteSQLException.class, exMsg);
- }
-}
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccSelectForUpdateQueryBasicTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccSelectForUpdateQueryBasicTest.java
new file mode 100644
index 0000000..5795b0a
--- /dev/null
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccSelectForUpdateQueryBasicTest.java
@@ -0,0 +1,789 @@
+/*
+ * 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.cache.mvcc;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.Callable;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import javax.cache.CacheException;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cache.query.FieldsQueryCursor;
+import org.apache.ignite.cache.query.SqlFieldsQuery;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.util.typedef.T2;
+import org.apache.ignite.internal.util.typedef.X;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.transactions.Transaction;
+import org.apache.ignite.transactions.TransactionConcurrency;
+import org.apache.ignite.transactions.TransactionIsolation;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static org.apache.ignite.cache.CacheMode.PARTITIONED;
+import static org.apache.ignite.cache.CacheMode.REPLICATED;
+import static org.apache.ignite.internal.processors.cache.index.AbstractSchemaSelfTest.connect;
+import static org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
+import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ;
+
+/**
+ *
+ */
+@RunWith(Parameterized.class)
+public class CacheMvccSelectForUpdateQueryBasicTest extends CacheMvccAbstractTest {
+ /** */
+ private static final int CACHE_SIZE = 100;
+
+ /** */
+ private static Random RAND = new Random();
+
+ /** */
+ @Parameterized.Parameter(0)
+ public CacheMode cacheMode;
+
+ /** */
+ @Parameterized.Parameter(1)
+ public int backups;
+
+ /** */
+ @Parameterized.Parameter(2)
+ public boolean fromClient;
+
+ /** */
+ @Parameterized.Parameter(3)
+ public boolean segmented;
+
+ /**
+ * @return Test parameters.
+ */
+ @Parameterized.Parameters(name = "cacheMode={0}, backups={1}, fromClient={2}, segmented={3}")
+ public static Collection parameters() {
+ return Arrays.asList(new Object[][] {
+ // cacheMode, backups, from client, segmented
+ { REPLICATED, 0, true, false},
+ { REPLICATED, 0, false, false},
+ { PARTITIONED, 0, true, false},
+ { PARTITIONED, 0, false, true},
+ { PARTITIONED, 1, true, true},
+ { PARTITIONED, 1, false, false},
+ { PARTITIONED, 2, true, false},
+ });
+ }
+
+ /** {@inheritDoc} */
+ @Override protected CacheMode cacheMode() {
+ return null;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ cleanPersistenceDir();
+
+ startGridsMultiThreaded(3);
+
+ client = true;
+
+ startGrid(3);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ super.afterTestsStopped();
+
+ super.afterTest(); // Check mvcc state and stop nodes
+ }
+
+ /** {@inheritDoc} */
+ @Override public void afterTest() throws Exception {
+ verifyOldVersionsCleaned();
+
+ verifyCoordinatorInternalState();
+
+ grid(3).destroyCache("SQL_PUBLIC_PERSON");
+ }
+
+ /** {@inheritDoc} */
+ @Override public void beforeTest() throws Exception {
+ Ignite client = grid(3);
+
+ CacheConfiguration<Object, Object> ccfg = new CacheConfiguration<>("dummy")
+ .setSqlSchema("PUBLIC").setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL);
+
+ // Create dummy cache as entry point.
+ client.getOrCreateCache(ccfg);
+
+ if (segmented) {
+ CacheConfiguration<Object, Object> seg = new CacheConfiguration<>("segmented*");
+
+ seg.setQueryParallelism(4);
+
+ client.addCacheConfiguration(seg);
+ }
+
+ // Create MVCC table and cache.
+ runSql(client, "CREATE TABLE person (id INT PRIMARY KEY, name VARCHAR, salary INT) " +
+ "WITH \"ATOMICITY=TRANSACTIONAL_SNAPSHOT, BACKUPS=" + backups +
+ (segmented ? ", TEMPLATE=segmented" : "") + "\"", false);
+
+ runSql(client, "CREATE INDEX salaryIdx ON person(salary)", false);
+
+ // Populate MVCC cache. Salaries 0, 10, 20, 30,..., 990.
+ for (int i = 0; i < CACHE_SIZE; i++)
+ runSql(client, "INSERT INTO person (id, name, salary) VALUES ("
+ + i + ", 'name" + i + "', " + i * 10 + ")", false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testSingleLock() throws Exception {
+ Ignite node = getNode();
+
+ String sql = "SELECT id, name FROM person WHERE salary = 100";
+ String sqlForUpdate = sql + " FOR UPDATE";
+
+ // Check SELECT FOR UPDATE.
+ try (Transaction tx = node.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ List<List<?>> res = runSql(node, sqlForUpdate, false).getAll();
+
+ assertEquals(1, res.size());
+
+ List<Integer> keys = new ArrayList<>();
+
+ for (List<?> r : res) {
+ assertEquals(2, r.size());
+
+ int key = (Integer)r.get(0);
+
+ assertEquals(10, key);
+
+ keys.add(key);
+ }
+
+ checkLocks(keys);
+
+ tx.commit();
+ }
+
+ checkLocks(Collections.emptyList());
+
+ // Check cached statement works in the same way.
+ try (Transaction tx = node.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ List<List<?>> res = runSql(node, sqlForUpdate, false).getAll();
+
+ assertEquals(1, res.size());
+
+ List<Integer> keys = new ArrayList<>();
+
+ for (List<?> r : res) {
+ assertEquals(2, r.size());
+
+ int key = (Integer)r.get(0);
+
+ assertEquals(10, key);
+
+ keys.add(key);
+ }
+
+ checkLocks(keys);
+
+ tx.rollback();
+
+ checkLocks(Collections.emptyList());
+ }
+
+ // Check cached statement doesn't lock keys without tx.
+ {
+ List<List<?>> res = runSql(node, sqlForUpdate, false).getAll();
+
+ assertEquals(1, res.size());
+
+ List<Integer> keys = new ArrayList<>();
+
+ for (List<?> r : res) {
+ assertEquals(2, r.size());
+
+ int key = (Integer)r.get(0);
+
+ assertEquals(10, key);
+
+ keys.add(key);
+ }
+
+ checkLocks(Collections.emptyList());
+ }
+
+ // Check cached statement locks keys in tx.
+ try (Transaction tx = node.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ List<List<?>> res = runSql(node, sqlForUpdate, false).getAll();
+
+ assertEquals(1, res.size());
+
+ List<Integer> keys = new ArrayList<>();
+
+ for (List<?> r : res) {
+ assertEquals(2, r.size());
+
+ int key = (Integer)r.get(0);
+
+ assertEquals(10, key);
+
+ keys.add(key);
+ }
+
+ // Run dummy DML.
+ runSql(node, "UPDATE Person SET name='test' WHERE id=" + keys.get(0), false) .getAll();
+
+ checkLocks(keys);
+
+ tx.commit();
+ }
+
+ checkLocks(Collections.emptyList());
+
+ // Check SFU and non-SFU selects are cached separately.
+ try (Transaction tx = node.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ List<List<?>> res = runSql(node, sql, false).getAll();
+
+ assertEquals(1, res.size());
+
+ for (List<?> r : res)
+ assertEquals(2, r.size());
+
+ checkLocks(Collections.emptyList());
+
+ tx.commit();
+ }
+
+ // Check SFU and non-SFU selects are cached separately.
+ {
+ List<List<?>> res = runSql(node, sql, false).getAll();
+
+ assertEquals(1, res.size());
+
+ for (List<?> r : res)
+ assertEquals(2, r.size());
+
+ checkLocks(Collections.emptyList());
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testSelectForUpdateLocal() throws Exception {
+ checkSelectForUpdate(true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testSelectForUpdateDistributed() throws Exception {
+ checkSelectForUpdate(false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ private void checkSelectForUpdate(boolean loc) throws Exception {
+ Ignite node = loc ? grid(0) : getNode();
+
+ String sql = "SELECT name, id FROM person WHERE MOD(salary, 3) = 0 ORDER BY id";
+ String sqlForUpdate = sql + " FOR UPDATE";
+
+ // Check statement doesn't lock keys without tx.
+ {
+ List<List<?>> res = runSql(node, sqlForUpdate, loc).getAll();
+
+ // Every third row should be fetched in distributed case.
+ assertTrue(loc ? !res.isEmpty() : res.size() >= CACHE_SIZE / 4 && res.size() <= CACHE_SIZE / 2);
+
+ List<Integer> keys = new ArrayList<>();
+
+ for (List<?> r : res) {
+ assertEquals(2, r.size());
+
+ keys.add((Integer)r.get(1));
+ }
+
+ checkLocks(Collections.emptyList());
+ }
+
+ // Check SELECT FOR UPDATE.
+ try (Transaction tx = node.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ List<List<?>> res = runSql(node, sqlForUpdate, loc).getAll();
+
+ // Every third row should be fetched in distributed case.
+ assertTrue(loc ? !res.isEmpty() : res.size() >= CACHE_SIZE / 4 && res.size() <= CACHE_SIZE / 2);
+
+ List<Integer> keys = new ArrayList<>();
+
+ for (List<?> r : res) {
+ assertEquals(2, r.size());
+
+ keys.add((Integer)r.get(1));
+ }
+
+ // Run dummy DML.
+ runSql(node, "UPDATE Person SET name='test' WHERE id=" + keys.get(0), loc) .getAll();
+
+ checkLocks(keys);
+
+ tx.rollback();
+
+ checkLocks(Collections.emptyList());
+ }
+
+ // Check cached statement doesn't lock keys without tx.
+ {
+ List<List<?>> res = runSql(node, sqlForUpdate, loc).getAll();
+
+ // Every third row should be fetched in distributed case.
+ assertTrue(loc ? !res.isEmpty() : res.size() >= CACHE_SIZE / 4 && res.size() <= CACHE_SIZE / 2);
+
+ List<Integer> keys = new ArrayList<>();
+
+ for (List<?> r : res) {
+ assertEquals(2, r.size());
+
+ keys.add((Integer)r.get(1));
+ }
+
+ checkLocks(Collections.emptyList());
+ }
+
+ try (Transaction tx = node.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ List<List<?>> res = runSql(node, sqlForUpdate, loc).getAll();
+
+ // Every third row should be fetched in distributed case.
+ assertTrue(loc ? !res.isEmpty() : res.size() >= CACHE_SIZE / 4 && res.size() <= CACHE_SIZE / 2);
+
+ List<Integer> keys = new ArrayList<>();
+
+ for (List<?> r : res) {
+ assertEquals(2, r.size());
+
+ keys.add((Integer)r.get(1));
+ }
+
+ // Run dummy DML.
+ runSql(node, "UPDATE Person SET name='test' WHERE id=" + keys.get(0), loc) .getAll();
+
+ checkLocks(keys);
+
+ tx.rollback();
+
+ checkLocks(Collections.emptyList());
+ }
+
+ // Check SFU and non-SFU selects are cached separately.
+ {
+ List<List<?>> res = runSql(node, sql, loc).getAll();
+
+ // Every third row should be fetched in distributed case.
+ assertTrue(loc ? !res.isEmpty() : res.size() >= CACHE_SIZE / 4 && res.size() <= CACHE_SIZE / 2);
+
+ for (List<?> r : res)
+ assertEquals(2, r.size());
+
+ checkLocks(Collections.emptyList());
+ }
+
+ // Check SFU and non-SFU selects are cached separately.
+ try (Transaction tx = node.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ List<List<?>> res = runSql(node, sql, loc).getAll();
+
+ // Every third row should be fetched in distributed case.
+ assertTrue(loc ? !res.isEmpty() : res.size() >= CACHE_SIZE / 4 && res.size() <= CACHE_SIZE / 2);
+
+ List<Integer> keys = new ArrayList<>();
+
+ for (List<?> r : res) {
+ assertEquals(2, r.size());
+
+ keys.add((Integer)r.get(1));
+ }
+
+ checkLocks(Collections.emptyList());
+
+ tx.rollback();
+ }
+
+ checkLocks(Collections.emptyList());
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testJdbcAutoCommitFalseStartsTx() throws Exception {
+ Ignite node = getNode();
+
+ try (Connection c = connect((IgniteEx)node)) {
+ c.setAutoCommit(false);
+
+ String sql = "SELECT name, id FROM person WHERE MOD(salary, 3) = 0 ORDER BY id";
+ String sqlForUpdate = sql + " FOR UPDATE";
+
+ List<Integer> keys = runJdbcSql(c, sqlForUpdate, 2, 2);
+
+ assertTrue(keys.size() >= CACHE_SIZE / 4 && keys.size() <= CACHE_SIZE / 2);
+
+ checkLocks(keys);
+
+ runJdbcSql(c, sql, 2, 2);
+
+ checkLocks(keys);
+
+ runJdbcSql(c, "COMMIT", -1, -1);
+
+ checkLocks(Collections.emptyList());
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testJdbcAutoCommitTrueKeysLocked() throws Exception {
+ Ignite node = getNode();
+
+ try (Connection c = connect((IgniteEx)node)) {
+ c.setAutoCommit(true);
+
+ runJdbcSql(c, "BEGIN", -1, -1);
+
+ String sql = "SELECT name, id FROM person WHERE MOD(salary, 3) = 0 ORDER BY id";
+ String sqlForUpdate = sql + " FOR UPDATE";
+
+ List<Integer> keys = runJdbcSql(c, sqlForUpdate, 2, 2);
+
+ assertTrue(keys.size() >= CACHE_SIZE / 4 && keys.size() <= CACHE_SIZE / 2);
+
+ checkLocks(keys);
+
+ runJdbcSql(c, sql, 2, 2);
+
+ checkLocks(keys);
+
+ runJdbcSql(c, "COMMIT", -1, -1);
+
+ checkLocks(Collections.emptyList());
+
+ // Should not be locked because we are out of tx.
+ runJdbcSql(c, sqlForUpdate, 2, 2);
+
+ checkLocks(Collections.emptyList());
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testSelectForUpdateLocalWithArgs() throws Exception {
+ checkSelectForUpdateWithArgs(true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testSelectForUpdateDistributedWithArgs() throws Exception {
+ checkSelectForUpdateWithArgs(false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ private void checkSelectForUpdateWithArgs(boolean loc) throws Exception {
+ Ignite node = loc ? grid(0) : getNode();
+
+ String sql = "SELECT name, id FROM person WHERE salary >= ? AND salary < ? ORDER BY id";
+ String sqlForUpdate = sql + " FOR UPDATE";
+
+ // Check cached statement doesn't lock keys without tx.
+ {
+ List<List<?>> res = runSql(node, sqlForUpdate, loc, 0, 200).getAll();
+
+ // Every third row should be fetched in distributed case.
+ assertTrue(loc ? !res.isEmpty() : res.size() == 20);
+
+ List<Integer> keys = new ArrayList<>();
+
+ for (List<?> r : res) {
+ assertEquals(2, r.size());
+
+ keys.add((Integer)r.get(1));
+ }
+
+ checkLocks(Collections.emptyList());
+ }
+
+ // Check SELECT FOR UPDATE.
+ try (Transaction tx = node.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ List<List<?>> res = runSql(node, sqlForUpdate, loc, 0, 200).getAll();
+
+ // Every third row should be fetched in distributed case.
+ assertTrue(loc ? !res.isEmpty() : res.size() == 20);
+
+ List<Integer> keys = new ArrayList<>();
+
+ for (List<?> r : res) {
+ assertEquals(2, r.size());
+
+ keys.add((Integer)r.get(1));
+ }
+
+ // Run dummy DML.
+ runSql(node, "UPDATE Person SET name='test' WHERE id=" + keys.get(0), loc) .getAll();
+
+ checkLocks(keys);
+
+ tx.rollback();
+
+ checkLocks(Collections.emptyList());
+ }
+
+ // Check cached statement doesn't lock keys without tx.
+ {
+ List<List<?>> res = runSql(node, sqlForUpdate, loc, 0, 200).getAll();
+
+ // Every third row should be fetched in distributed case.
+ assertTrue(loc ? !res.isEmpty() : res.size() == 20);
+
+ List<Integer> keys = new ArrayList<>();
+
+ for (List<?> r : res) {
+ assertEquals(2, r.size());
+
+ keys.add((Integer)r.get(1));
+ }
+
+ checkLocks(Collections.emptyList());
+ }
+
+ try (Transaction tx = node.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ List<List<?>> res = runSql(node, sqlForUpdate, loc, 0, 200).getAll();
+
+ // Every third row should be fetched in distributed case.
+ assertTrue(loc ? !res.isEmpty() : res.size() == 20);
+
+ List<Integer> keys = new ArrayList<>();
+
+ for (List<?> r : res) {
+ assertEquals(2, r.size());
+
+ keys.add((Integer)r.get(1));
+ }
+
+ // Run dummy DML.
+ runSql(node, "UPDATE Person SET name='test' WHERE id=" + keys.get(0), loc) .getAll();
+
+ checkLocks(keys);
+
+ tx.rollback();
+
+ checkLocks(Collections.emptyList());
+ }
+
+ // Check SFU and non-SFU selects are cached separately.
+ {
+ List<List<?>> res = runSql(node, sql, loc, 0, 200).getAll();
+
+ // Every third row should be fetched in distributed case.
+ assertTrue(loc ? !res.isEmpty() : res.size() == 20);
+
+ for (List<?> r : res)
+ assertEquals(2, r.size());
+
+ checkLocks(Collections.emptyList());
+ }
+
+ // Check SFU and non-SFU selects are cached separately.
+ try (Transaction tx = node.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ List<List<?>> res = runSql(node, sql, loc, 0, 200).getAll();
+
+ // Every third row should be fetched in distributed case.
+ assertTrue(loc ? !res.isEmpty() : res.size() == 20);
+
+ List<Integer> keys = new ArrayList<>();
+
+ for (List<?> r : res) {
+ assertEquals(2, r.size());
+
+ keys.add((Integer)r.get(1));
+ }
+
+ checkLocks(Collections.emptyList());
+
+ tx.rollback();
+ }
+
+ checkLocks(Collections.emptyList());
+ }
+
+ /**
+ * Check that an attempt to get a lock on any key from given list fails by timeout.
+ *
+ * @param lockedKeys Keys to check.
+ * @throws Exception if failed.
+ */
+ @SuppressWarnings({"unchecked"})
+ private void checkLocks(List<Integer> lockedKeys) throws Exception {
+ List<Integer> allKeys = IntStream.range(0, CACHE_SIZE).boxed().collect(Collectors.toList());
+ List<Integer> nonLockedKeys = new ArrayList<>(allKeys);
+
+ nonLockedKeys.removeAll(lockedKeys);
+
+ List<Ignite> nodes = Ignition.allGrids();
+
+ Ignite node = nodes.get(RAND.nextInt(nodes.size()));
+
+ List<T2<Integer, IgniteInternalFuture>> calls = new ArrayList<>();
+
+ for (int key : allKeys) {
+ calls.add(new T2<>(key, GridTestUtils.runAsync(new Callable<Void>() {
+ /** {@inheritDoc} */
+ @Override public Void call() {
+ try (Transaction tx = node.transactions().txStart(TransactionConcurrency.PESSIMISTIC,
+ TransactionIsolation.REPEATABLE_READ)) {
+
+ // TODO uncomment to reproduce "https://issues.apache.org/jira/browse/IGNITE-11349"
+// if (RAND.nextBoolean()) {
+// node.cache("dummy").query(new SqlFieldsQuery("update person" +
+// " set name='check' where id = " + key)
+// .setTimeout(1, TimeUnit.SECONDS))
+// .getAll();
+// }
+// else {
+ node.cache("dummy").query(new SqlFieldsQuery("SELECT * FROM person WHERE id="+ key +
+ " FOR UPDATE")
+ .setTimeout(1, TimeUnit.SECONDS))
+ .getAll();
+// }
+
+ tx.rollback();
+
+ return null;
+ }
+ }
+ })));
+ }
+
+ for (T2<Integer, IgniteInternalFuture> pair : calls) {
+ if (nonLockedKeys.contains(pair.getKey())) {
+ try {
+ pair.getValue().get(TX_TIMEOUT);
+ }
+ catch (Exception e) {
+ if (e.getMessage() != null && e.getMessage().contains("Failed to acquire lock within provided timeout"))
+ throw new Exception("Key is locked, though it shouldn't be. Key: " + pair.getKey(), e);
+
+ throw e;
+ }
+ }
+ else {
+ try {
+ pair.getValue().get();
+
+ fail("Key is not locked: " + pair.getKey());
+ }
+ catch (Exception e) {
+ CacheException e0 = X.cause(e, CacheException.class);
+
+ assert e0 != null;
+
+ assert e0.getMessage() != null &&
+ e0.getMessage().contains("Failed to acquire lock within provided timeout") :
+ X.getFullStackTrace(e);
+ }
+ }
+ }
+ }
+
+ /**
+ * @return Node to run query from.
+ */
+ private Ignite getNode() {
+ Ignite node = fromClient ? grid(3) : grid(RAND.nextInt(2));
+
+ assert fromClient == node.cluster().localNode().isClient();
+
+ return node;
+ }
+
+ /**
+ * @param node Node.
+ * @param sql Sql string.
+ * @return Result.
+ */
+ private FieldsQueryCursor<List<?>> runSql(Ignite node, String sql, boolean local, Object... args) {
+ return node.cache("dummy").query(new SqlFieldsQuery(sql).setLocal(local).setArgs(args));
+ }
+
+ /**
+ * @param c Connection.
+ * @param sql Sql string.
+ * @param keyIdx Index of key field in result set.
+ * @param colCnt Expected columns count in result set.
+ */
+ private List<Integer> runJdbcSql(Connection c, String sql, int keyIdx, int colCnt) throws SQLException {
+ try (Statement stmt = c.createStatement()) {
+ stmt.execute(sql);
+
+ ResultSet rs = stmt.getResultSet();
+
+ if (rs == null) {
+ assertEquals(keyIdx, -1);
+
+ return null;
+ }
+
+ List<Integer> res = new ArrayList<>();
+
+ while (rs.next()) {
+ assertEquals(colCnt, rs.getMetaData().getColumnCount());
+
+ res.add(rs.getInt(keyIdx));
+ }
+
+ return res;
+ }
+ }
+}
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccSelectForUpdateQueryTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccSelectForUpdateQueryTest.java
new file mode 100644
index 0000000..37526e5
--- /dev/null
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccSelectForUpdateQueryTest.java
@@ -0,0 +1,184 @@
+/*
+ * 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.cache.mvcc;
+
+import java.sql.Connection;
+import java.util.List;
+import java.util.concurrent.Callable;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.cache.query.SqlFieldsQuery;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.processors.query.IgniteSQLException;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.transactions.Transaction;
+import org.apache.ignite.transactions.TransactionConcurrency;
+import org.apache.ignite.transactions.TransactionIsolation;
+import org.junit.Test;
+
+import static org.apache.ignite.internal.processors.cache.index.AbstractSchemaSelfTest.connect;
+import static org.apache.ignite.internal.processors.cache.index.AbstractSchemaSelfTest.execute;
+
+/**
+ * Test for {@code SELECT FOR UPDATE} queries.
+ */
+public class CacheMvccSelectForUpdateQueryTest extends GridCommonAbstractTest {
+ /** */
+ private static final int CACHE_SIZE = 10;
+
+ /** {@inheritDoc} */
+ @SuppressWarnings("unchecked")
+ @Override protected void beforeTestsStarted() throws Exception {
+ Ignite grid = startGridsMultiThreaded(2);
+
+ try (Connection c = connect((IgniteEx)grid)) {
+ c.setAutoCommit(false);
+
+ execute(c, "create table person (id int primary key, firstName varchar, lastName varchar) " +
+ "with \"atomicity=transactional_snapshot,cache_name=Person\"");
+
+ execute(c, "create table person_nonMvcc (id int primary key, firstName varchar, lastName varchar) " +
+ "with \"atomicity=transactional,cache_name=Person_nonMvcc\"");
+
+ try (Transaction tx = grid(0).transactions().txStart(TransactionConcurrency.PESSIMISTIC,
+ TransactionIsolation.REPEATABLE_READ)) {
+
+ for (int i = 1; i <= CACHE_SIZE; i++) {
+ execute(c, "insert into person(id, firstName, lastName) " +
+ "values(" + i + ",'firstName" + i + "','lastName" + i + "')");
+ }
+
+ tx.commit();
+ }
+ }
+ }
+
+ /**
+ *
+ */
+ @Test
+ public void testSelectForUpdateWithUnion() {
+ assertQueryThrows("select id from person union select 1 for update",
+ "SELECT UNION FOR UPDATE is not supported.");
+ }
+
+ /**
+ *
+ */
+ @Test
+ public void testSelectForUpdateWithMultipleStatements() {
+ assertQueryThrows("select id from person for update; select firstName from person for update",
+ "Multiple statements queries are not supported");
+ }
+
+ /**
+ *
+ */
+ @Test
+ public void testSelectForUpdateWithJoin() {
+ assertQueryThrows("select p1.id from person p1 join person p2 on p1.id = p2.id for update",
+ "SELECT FOR UPDATE with joins is not supported.");
+ }
+
+ /**
+ *
+ */
+ @Test
+ public void testSelectForUpdateWithLimit() {
+ assertQueryThrows("select id from person limit 0,5 for update",
+ "LIMIT/OFFSET clauses are not supported for SELECT FOR UPDATE.");
+ }
+
+ /**
+ *
+ */
+ @Test
+ public void testSelectForUpdateWithOffset() {
+ assertQueryThrows("select id from person offset 10 for update",
+ "LIMIT/OFFSET clauses are not supported for SELECT FOR UPDATE.");
+ }
+
+ /**
+ *
+ */
+ @Test
+ public void testSelectForUpdateWithDistinct() {
+ assertQueryThrows("select distinct firstName from PERSON for update",
+ "DISTINCT clause is not supported for SELECT FOR UPDATE.");
+ }
+
+ /**
+ *
+ */
+ @Test
+ public void testSelectForUpdateWithSubQuery() {
+ assertQueryThrows("select id, firstName from PERSON where id = (SELECT COUNT(*) FROM person) for update",
+ "Sub queries are not supported for SELECT FOR UPDATE.");
+ }
+
+ /**
+ *
+ */
+ @Test
+ public void testSelectForUpdateNonMvccCache() {
+ assertQueryThrows("select id, firstName from person_nonMvcc for update",
+ "SELECT FOR UPDATE query requires transactional cache with MVCC enabled.");
+ }
+
+ /**
+ *
+ */
+ @Test
+ public void testSelectForUpdateWithGroupings() {
+ assertQueryThrows("select count(*) from person for update",
+ "SELECT FOR UPDATE with aggregates and/or GROUP BY is not supported.");
+
+ assertQueryThrows("select lastName, count(*) from person group by lastName for update",
+ "SELECT FOR UPDATE with aggregates and/or GROUP BY is not supported.");
+ }
+
+ /**
+ * Test that query throws exception with expected message.
+ * @param qry SQL.
+ * @param exMsg Expected message.
+ */
+ private void assertQueryThrows(String qry, String exMsg) {
+ assertQueryThrows(qry, exMsg, false);
+
+ assertQueryThrows(qry, exMsg, true);
+ }
+
+ /**
+ * Test that query throws exception with expected message.
+ * @param qry SQL.
+ * @param exMsg Expected message.
+ * @param loc Local query flag.
+ */
+ @SuppressWarnings("ThrowableNotThrown")
+ private void assertQueryThrows(String qry, String exMsg, boolean loc) {
+ Ignite node = grid(0);
+
+ GridTestUtils.assertThrows(null, new Callable<Object>() {
+ @Override public Object call() {
+ List r = node.cache("Person").query(new SqlFieldsQuery(qry).setLocal(loc)).getAll();
+
+ return r;
+ }
+ }, IgniteSQLException.class, exMsg);
+ }
+}
diff --git a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite2.java b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite2.java
index 5856ef7..9e906fb 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite2.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite2.java
@@ -52,6 +52,7 @@ import org.apache.ignite.internal.processors.query.IgniteCacheGroupsSqlSegmented
import org.apache.ignite.internal.processors.query.IgniteCacheGroupsSqlSegmentedIndexSelfTest;
import org.apache.ignite.internal.processors.query.IgniteSqlCreateTableTemplateTest;
import org.apache.ignite.internal.processors.query.LocalQueryLazyTest;
+import org.apache.ignite.internal.processors.query.LongRunningQueryTest;
import org.apache.ignite.internal.processors.query.SqlLocalQueryConnectionAndStatementTest;
import org.apache.ignite.internal.processors.query.h2.CacheQueryEntityWithDateTimeApiFieldsTest;
import org.apache.ignite.internal.processors.query.h2.DmlStatementsProcessorTest;
@@ -139,6 +140,8 @@ import org.junit.runners.Suite;
IgniteSqlCreateTableTemplateTest.class,
LocalQueryLazyTest.class,
+
+ LongRunningQueryTest.class,
})
public class IgniteBinaryCacheQueryTestSuite2 {
}
diff --git a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheMvccSqlTestSuite.java b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheMvccSqlTestSuite.java
index e7e370b..ef2d78a 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheMvccSqlTestSuite.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheMvccSqlTestSuite.java
@@ -44,18 +44,18 @@ import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccDmlSimpleTest;
import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccIteratorWithConcurrentJdbcTransactionTest;
import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccLocalEntriesWithConcurrentJdbcTransactionTest;
import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccPartitionedBackupsTest;
-import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccPartitionedSelectForUpdateQueryTest;
import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccPartitionedSqlCoordinatorFailoverTest;
import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccPartitionedSqlQueriesTest;
import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccPartitionedSqlTxQueriesTest;
import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccPartitionedSqlTxQueriesWithReducerTest;
import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccReplicatedBackupsTest;
-import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccReplicatedSelectForUpdateQueryTest;
import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccReplicatedSqlCoordinatorFailoverTest;
import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccReplicatedSqlQueriesTest;
import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccReplicatedSqlTxQueriesTest;
import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccReplicatedSqlTxQueriesWithReducerTest;
import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccScanQueryWithConcurrentJdbcTransactionTest;
+import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccSelectForUpdateQueryTest;
+import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccSelectForUpdateQueryBasicTest;
import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccSizeTest;
import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccSizeWithConcurrentJdbcTransactionTest;
import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccSqlConfigurationValidationTest;
@@ -117,8 +117,9 @@ import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL_SNAPSHOT;
CacheMvccPartitionedSqlTxQueriesWithReducerTest.class,
CacheMvccReplicatedSqlTxQueriesWithReducerTest.class,
- CacheMvccPartitionedSelectForUpdateQueryTest.class,
- CacheMvccReplicatedSelectForUpdateQueryTest.class,
+
+ CacheMvccSelectForUpdateQueryBasicTest.class,
+ CacheMvccSelectForUpdateQueryTest.class,
// Failover tests.
CacheMvccPartitionedBackupsTest.class,