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/02/05 14:58:58 UTC
[ignite] branch master updated: IGNITE-11202: SQL: Moved partition
reservation logic out of GridMapQueryExecutor to allow for it's reuse in
future. This closes #6023.
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 299f557 IGNITE-11202: SQL: Moved partition reservation logic out of GridMapQueryExecutor to allow for it's reuse in future. This closes #6023.
299f557 is described below
commit 299f557d9b0b7cbc4f7584d28be302056c6fb46c
Author: devozerov <vo...@gridgain.com>
AuthorDate: Tue Feb 5 17:58:46 2019 +0300
IGNITE-11202: SQL: Moved partition reservation logic out of GridMapQueryExecutor to allow for it's reuse in future. This closes #6023.
---
.../processors/query/h2/IgniteH2Indexing.java | 15 +-
.../processors/query/h2/opt/QueryContext.java | 17 +-
.../query/h2/twostep/GridMapQueryExecutor.java | 298 ++-----------------
.../query/h2/twostep/MapReplicatedReservation.java | 38 ---
.../query/h2/twostep/PartitionReservation.java | 92 ++++++
...vationKey.java => PartitionReservationKey.java} | 8 +-
.../h2/twostep/PartitionReservationManager.java | 319 +++++++++++++++++++++
.../h2/twostep/RetryCauseMessageSelfTest.java | 34 ++-
8 files changed, 482 insertions(+), 339 deletions(-)
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 d3aeedc..9ee056f 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
@@ -96,6 +96,7 @@ import org.apache.ignite.internal.processors.query.h2.affinity.PartitionExtracto
import org.apache.ignite.internal.processors.query.h2.affinity.H2PartitionResolver;
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.twostep.PartitionReservationManager;
import org.apache.ignite.internal.sql.optimizer.affinity.PartitionResult;
import org.apache.ignite.internal.processors.query.h2.database.H2TreeClientIndex;
import org.apache.ignite.internal.processors.query.h2.database.H2TreeIndex;
@@ -253,6 +254,9 @@ public class IgniteH2Indexing implements GridQueryIndexing {
/** */
private DdlStatementsProcessor ddlProc;
+ /** Partition reservation manager. */
+ private PartitionReservationManager partReservationMgr;
+
/** Partition extractor. */
private PartitionExtractor partExtractor;
@@ -2400,6 +2404,8 @@ public class IgniteH2Indexing implements GridQueryIndexing {
this.ctx = ctx;
+ partReservationMgr = new PartitionReservationManager(ctx);
+
connMgr = new ConnectionManager(ctx);
schemaMgr = new SchemaManager(ctx, connMgr);
@@ -2607,7 +2613,7 @@ public class IgniteH2Indexing implements GridQueryIndexing {
String cacheName = cacheInfo.name();
- mapQryExec.onCacheStop(cacheName);
+ partReservationMgr.onCacheStop(cacheName);
dmlProc.onCacheStop(cacheName);
// Drop schema (needs to be called after callback to DML processor because the latter depends on schema).
@@ -2747,6 +2753,13 @@ public class IgniteH2Indexing implements GridQueryIndexing {
}
/**
+ * @return Partition reservation manager.
+ */
+ public PartitionReservationManager partitionReservationManager() {
+ return partReservationMgr;
+ }
+
+ /**
* Collect cache identifiers from two-step query.
*
* @param mainCacheId Id of main cache.
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/QueryContext.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/QueryContext.java
index e66ec7f..ca0c940 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/QueryContext.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/QueryContext.java
@@ -17,17 +17,14 @@
package org.apache.ignite.internal.processors.query.h2.opt;
-import org.apache.ignite.internal.processors.cache.distributed.dht.GridReservable;
import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
import org.apache.ignite.internal.processors.query.h2.opt.join.DistributedJoinContext;
import org.apache.ignite.internal.processors.query.h2.twostep.MapQueryLazyWorker;
-import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.processors.query.h2.twostep.PartitionReservation;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.spi.indexing.IndexingQueryFilter;
import org.jetbrains.annotations.Nullable;
-import java.util.List;
-
/**
* Thread local SQL query context which is intended to be accessible from everywhere.
*/
@@ -45,7 +42,7 @@ public class QueryContext {
private final MvccSnapshot mvccSnapshot;
/** */
- private final List<GridReservable> reservations;
+ private final PartitionReservation reservations;
/** */
private MapQueryLazyWorker lazyWorker;
@@ -64,7 +61,7 @@ public class QueryContext {
@Nullable IndexingQueryFilter filter,
@Nullable DistributedJoinContext distributedJoinCtx,
@Nullable MvccSnapshot mvccSnapshot,
- @Nullable List<GridReservable> reservations
+ @Nullable PartitionReservation reservations
) {
this.segment = segment;
this.filter = filter;
@@ -102,12 +99,8 @@ public class QueryContext {
if (distributedJoinCtx != null)
distributedJoinCtx.cancel();
- List<GridReservable> r = reservations;
-
- if (!nodeStop && !F.isEmpty(r)) {
- for (int i = 0; i < r.size(); i++)
- r.get(i).release();
- }
+ if (!nodeStop && reservations != null)
+ reservations.release();
}
/**
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 7ecf2c4..9b35f04 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
@@ -39,7 +39,6 @@ import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.IgniteSystemProperties;
-import org.apache.ignite.cache.PartitionLossPolicy;
import org.apache.ignite.cache.query.QueryCancelledException;
import org.apache.ignite.cache.query.SqlFieldsQuery;
import org.apache.ignite.cluster.ClusterNode;
@@ -53,15 +52,10 @@ 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.CacheInvalidStateException;
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.distributed.dht.GridReservable;
-import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition;
-import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState;
-import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionsReservation;
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;
@@ -87,7 +81,6 @@ import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2DmlRespo
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.CI1;
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;
@@ -102,13 +95,8 @@ import org.h2.value.Value;
import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_SQL_FORCE_LAZY_RESULT_SET;
-import static org.apache.ignite.cache.PartitionLossPolicy.READ_ONLY_SAFE;
-import static org.apache.ignite.cache.PartitionLossPolicy.READ_WRITE_SAFE;
import static org.apache.ignite.events.EventType.EVT_CACHE_QUERY_EXECUTED;
import static org.apache.ignite.internal.managers.communication.GridIoPolicy.QUERY_POOL;
-import static org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion.NONE;
-import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.LOST;
-import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.OWNING;
import static org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2QueryRequest.isDataPageScanEnabled;
import static org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2ValueMessageFactory.toMessages;
@@ -139,9 +127,6 @@ public class GridMapQueryExecutor {
/** */
private final GridSpinBusyLock busyLock;
- /** */
- private final ConcurrentMap<MapReservationKey, GridReservable> reservations = new ConcurrentHashMap<>();
-
/** Lazy workers. */
private final ConcurrentHashMap<MapQueryLazyWorkerKey, MapQueryLazyWorker> lazyWorkers = new ConcurrentHashMap<>();
@@ -301,229 +286,6 @@ public class GridMapQueryExecutor {
}
/**
- * @param cctx Cache context.
- * @param p Partition ID.
- * @return Partition.
- */
- private GridDhtLocalPartition partition(GridCacheContext<?, ?> cctx, int p) {
- return cctx.topology().localPartition(p, NONE, false);
- }
-
- /**
- * @param cacheIds Cache IDs.
- * @param topVer Topology version.
- * @param explicitParts Explicit partitions list.
- * @param reserved Reserved list.
- * @param nodeId Node ID.
- * @param reqId Request ID.
- * @return String which is null in case of success or with causeMessage if failed
- * @throws IgniteCheckedException If failed.
- */
- private String reservePartitions(
- @Nullable List<Integer> cacheIds,
- AffinityTopologyVersion topVer,
- final int[] explicitParts,
- List<GridReservable> reserved,
- UUID nodeId,
- long reqId
- ) throws IgniteCheckedException {
- assert topVer != null;
-
- if (F.isEmpty(cacheIds))
- return null;
-
- Collection<Integer> partIds;
-
- if (explicitParts == null)
- partIds = null;
- else if (explicitParts.length == 0)
- partIds = Collections.emptyList();
- else {
- partIds = new ArrayList<>(explicitParts.length);
-
- for (int explicitPart : explicitParts)
- partIds.add(explicitPart);
- }
-
- for (int i = 0; i < cacheIds.size(); i++) {
- GridCacheContext<?, ?> cctx = ctx.cache().context().cacheContext(cacheIds.get(i));
-
- // Cache was not found, probably was not deployed yet.
- if (cctx == null) {
- return String.format("Failed to reserve partitions for query (cache is not found on " +
- "local node) [localNodeId=%s, rmtNodeId=%s, reqId=%s, affTopVer=%s, cacheId=%s]",
- ctx.localNodeId(), nodeId, reqId, topVer, cacheIds.get(i));
- }
-
- if (cctx.isLocal() || !cctx.rebalanceEnabled())
- continue;
-
- // For replicated cache topology version does not make sense.
- final MapReservationKey grpKey = new MapReservationKey(cctx.name(), cctx.isReplicated() ? null : topVer);
-
- GridReservable r = reservations.get(grpKey);
-
- if (explicitParts == null && r != null) { // Try to reserve group partition if any and no explicits.
- if (r != MapReplicatedReservation.INSTANCE) {
- if (!r.reserve())
- return String.format("Failed to reserve partitions for query (group " +
- "reservation failed) [localNodeId=%s, rmtNodeId=%s, reqId=%s, affTopVer=%s, cacheId=%s, " +
- "cacheName=%s]",ctx.localNodeId(), nodeId, reqId, topVer, cacheIds.get(i), cctx.name());
-
- reserved.add(r);
- }
- }
- else { // Try to reserve partitions one by one.
- int partsCnt = cctx.affinity().partitions();
-
- if (cctx.isReplicated()) { // Check all the partitions are in owning state for replicated cache.
- if (r == null) { // Check only once.
- for (int p = 0; p < partsCnt; p++) {
- GridDhtLocalPartition part = partition(cctx, p);
-
- // We don't need to reserve partitions because they will not be evicted in replicated caches.
- GridDhtPartitionState partState = part != null ? part.state() : null;
-
- if (partState != OWNING)
- return String.format("Failed to reserve partitions for query " +
- "(partition of REPLICATED cache is not in OWNING state) [" +
- "localNodeId=%s, rmtNodeId=%s, reqId=%s, affTopVer=%s, cacheId=%s, cacheName=%s, " +
- "part=%s, partFound=%s, partState=%s]",
- ctx.localNodeId(),
- nodeId,
- reqId,
- topVer,
- cacheIds.get(i),
- cctx.name(),
- p,
- (part != null),
- partState
- );
- }
-
- // Mark that we checked this replicated cache.
- reservations.putIfAbsent(grpKey, MapReplicatedReservation.INSTANCE);
- }
- }
- else { // Reserve primary partitions for partitioned cache (if no explicit given).
- if (explicitParts == null)
- partIds = cctx.affinity().primaryPartitions(ctx.localNodeId(), topVer);
-
- int reservedCnt = 0;
-
- for (int partId : partIds) {
- GridDhtLocalPartition part = partition(cctx, partId);
-
- GridDhtPartitionState partState = part != null ? part.state() : null;
-
- if (partState != OWNING) {
- if (partState == LOST)
- ignoreLostPartitionIfPossible(cctx, part);
- else {
- return String.format("Failed to reserve partitions for query " +
- "(partition of PARTITIONED cache is not found or not in OWNING state) [" +
- "localNodeId=%s, rmtNodeId=%s, reqId=%s, affTopVer=%s, cacheId=%s, " +
- "cacheName=%s, part=%s, partFound=%s, partState=%s]",
- ctx.localNodeId(),
- nodeId,
- reqId,
- topVer,
- cacheIds.get(i),
- cctx.name(),
- partId,
- (part != null),
- partState
- );
- }
- }
-
- if (!part.reserve()) {
- return String.format("Failed to reserve partitions for query " +
- "(partition of PARTITIONED cache cannot be reserved) [" +
- "localNodeId=%s, rmtNodeId=%s, reqId=%s, affTopVer=%s, cacheId=%s, " +
- "cacheName=%s, part=%s, partFound=%s, partState=%s]",
- ctx.localNodeId(),
- nodeId,
- reqId,
- topVer,
- cacheIds.get(i),
- cctx.name(),
- partId,
- true,
- partState
- );
- }
-
- reserved.add(part);
-
- reservedCnt++;
-
- // Double check that we are still in owning state and partition contents are not cleared.
- partState = part.state();
-
- if (partState != OWNING) {
- if (partState == LOST)
- ignoreLostPartitionIfPossible(cctx, part);
- else {
- return String.format("Failed to reserve partitions for query " +
- "(partition of PARTITIONED cache is not in OWNING state after reservation) [" +
- "localNodeId=%s, rmtNodeId=%s, reqId=%s, affTopVer=%s, cacheId=%s, " +
- "cacheName=%s, part=%s, partState=%s]",
- ctx.localNodeId(),
- nodeId,
- reqId,
- topVer,
- cacheIds.get(i),
- cctx.name(),
- partId,
- partState
- );
- }
- }
- }
-
- if (explicitParts == null && reservedCnt > 0) {
- // We reserved all the primary partitions for cache, attempt to add group reservation.
- GridDhtPartitionsReservation grp = new GridDhtPartitionsReservation(topVer, cctx, "SQL");
-
- if (grp.register(reserved.subList(reserved.size() - reservedCnt, reserved.size()))) {
- if (reservations.putIfAbsent(grpKey, grp) != null)
- throw new IllegalStateException("Reservation already exists.");
-
- grp.onPublish(new CI1<GridDhtPartitionsReservation>() {
- @Override public void apply(GridDhtPartitionsReservation r) {
- reservations.remove(grpKey, r);
- }
- });
- }
- }
- }
- }
- }
-
- return null;
- }
-
- /**
- * Decide whether to ignore or proceed with lost partition.
- *
- * @param cctx Cache context.
- * @param part Partition.
- * @throws IgniteCheckedException If failed.
- */
- private static void ignoreLostPartitionIfPossible(GridCacheContext cctx, GridDhtLocalPartition part)
- throws IgniteCheckedException {
- PartitionLossPolicy plc = cctx.config().getPartitionLossPolicy();
-
- if (plc != null) {
- if (plc == READ_ONLY_SAFE || plc == READ_WRITE_SAFE) {
- throw new CacheInvalidStateException("Failed to execute query because cache partition has been " +
- "lost [cacheName=" + cctx.name() + ", part=" + part + ']');
- }
- }
- }
-
- /**
* @param node Node.
* @param req Query request.
*/
@@ -822,21 +584,27 @@ public class GridMapQueryExecutor {
MapQueryResults qr = null;
- List<GridReservable> reserved = new ArrayList<>();
+ 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) {
// Reserve primary for topology version or explicit partitions.
- String err = reservePartitions(cacheIds, topVer, parts, reserved, node.id(), reqId);
+ reserved = h2.partitionReservationManager().reservePartitions(
+ cacheIds,
+ topVer,
+ parts,
+ node.id(),
+ reqId
+ );
- if (!F.isEmpty(err)) {
+ if (reserved.failed()) {
// Unregister lazy worker because re-try may never reach this node again.
if (lazy)
stopAndUnregisterCurrentLazyWorker();
- sendRetry(node, reqId, segmentId, err);
+ sendRetry(node, reqId, segmentId, reserved.error());
return;
}
@@ -868,8 +636,9 @@ public class GridMapQueryExecutor {
distributedJoinCtx,
mvccSnapshot,
reserved
- )
- .lazyWorker(worker);
+ );
+
+ qctx.lazyWorker(worker);
Connection conn = h2.connections().connectionForThread().connection(schemaName);
@@ -1054,11 +823,8 @@ public class GridMapQueryExecutor {
}
}
finally {
- if (reserved != null) {
- // Release reserved partitions.
- for (int i = 0; i < reserved.size(); i++)
- reserved.get(i).release();
- }
+ if (reserved != null)
+ reserved.release();
}
}
@@ -1097,20 +863,26 @@ public class GridMapQueryExecutor {
AffinityTopologyVersion topVer = req.topologyVersion();
- List<GridReservable> reserved = new ArrayList<>();
+ PartitionReservation reserved = null;
MapNodeResults nodeResults = resultsForNode(node.id());
try {
- String err = reservePartitions(cacheIds, topVer, parts, reserved, node.id(), reqId);
-
- if (!F.isEmpty(err)) {
+ reserved = h2.partitionReservationManager().reservePartitions(
+ cacheIds,
+ topVer,
+ parts,
+ node.id(),
+ reqId
+ );
+
+ if (reserved.failed()) {
U.error(log, "Failed to reserve partitions for DML request. [localNodeId=" + ctx.localNodeId() +
", nodeId=" + node.id() + ", reqId=" + req.requestId() + ", cacheIds=" + cacheIds +
", topVer=" + topVer + ", parts=" + Arrays.toString(parts) + ']');
sendUpdateResponse(node, reqId, null,
- "Failed to reserve partitions for DML request. " + err);
+ "Failed to reserve partitions for DML request. " + reserved.error());
return;
}
@@ -1173,11 +945,8 @@ public class GridMapQueryExecutor {
sendUpdateResponse(node, reqId, null, e.getMessage());
}
finally {
- if (!F.isEmpty(reserved)) {
- // Release reserved partitions.
- for (int i = 0; i < reserved.size(); i++)
- reserved.get(i).release();
- }
+ if (reserved != null)
+ reserved.release();
nodeResults.removeUpdate(reqId);
}
@@ -1394,17 +1163,6 @@ public class GridMapQueryExecutor {
}
/**
- * @param cacheName Cache name.
- */
- public void onCacheStop(String cacheName) {
- // Drop group reservations.
- for (MapReservationKey grpKey : reservations.keySet()) {
- if (F.eq(grpKey.cacheName(), cacheName))
- reservations.remove(grpKey);
- }
- }
-
- /**
* Unregister lazy worker if needed (i.e. if we are currently in lazy worker thread).
*/
public void stopAndUnregisterCurrentLazyWorker() {
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/MapReplicatedReservation.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/MapReplicatedReservation.java
deleted file mode 100644
index dd8237b..0000000
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/MapReplicatedReservation.java
+++ /dev/null
@@ -1,38 +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.twostep;
-
-import org.apache.ignite.internal.processors.cache.distributed.dht.GridReservable;
-
-/**
- * Mapper fake reservation object for replicated caches.
- */
-class MapReplicatedReservation implements GridReservable {
- /** */
- static final MapReplicatedReservation INSTANCE = new MapReplicatedReservation();
-
- /** {@inheritDoc} */
- @Override public boolean reserve() {
- throw new IllegalStateException();
- }
-
- /** {@inheritDoc} */
- @Override public void release() {
- throw new IllegalStateException();
- }
-}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/PartitionReservation.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/PartitionReservation.java
new file mode 100644
index 0000000..a3d3167
--- /dev/null
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/PartitionReservation.java
@@ -0,0 +1,92 @@
+/*
+ * 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.twostep;
+
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridReservable;
+import org.jetbrains.annotations.Nullable;
+
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Partition reservation for specific query.
+ */
+public class PartitionReservation {
+ /** Reserved partitions. */
+ private final List<GridReservable> reserved;
+
+ /** Error message. */
+ private final String err;
+
+ /** Release guard. */
+ private final AtomicBoolean releaseGuard = new AtomicBoolean();
+
+ /**
+ * Constructor for successfull reservation.
+ *
+ * @param reserved Reserved partitions.
+ */
+ public PartitionReservation(List<GridReservable> reserved) {
+ this(reserved, null);
+ }
+
+ /**
+ * Constructor for failed reservation.
+ *
+ * @param err Error message.
+ */
+ public PartitionReservation(String err) {
+ this(null, err);
+ }
+
+ /**
+ * Base constructor.
+ *
+ * @param reserved Reserved partitions.
+ * @param err Error message.
+ */
+ private PartitionReservation(@Nullable List<GridReservable> reserved, @Nullable String err) {
+ this.reserved = reserved;
+ this.err = err;
+ }
+
+ /**
+ * @return Error message (if any).
+ */
+ @Nullable public String error() {
+ return err;
+ }
+
+ /**
+ * @return {@code True} if reservation failed.
+ */
+ public boolean failed() {
+ return err != null;
+ }
+
+ /**
+ * Release partitions.
+ */
+ public void release() {
+ if (!releaseGuard.compareAndSet(false, true))
+ return;
+
+ for (int i = 0; i < reserved.size(); i++)
+ reserved.get(i).release();
+ }
+}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/MapReservationKey.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/PartitionReservationKey.java
similarity index 89%
rename from modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/MapReservationKey.java
rename to modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/PartitionReservationKey.java
index 9d2d7ba..b2ce082 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/MapReservationKey.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/PartitionReservationKey.java
@@ -21,9 +21,9 @@ import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.util.typedef.F;
/**
- * Mapper reservation key.
+ * Partition reservation key.
*/
-public class MapReservationKey {
+public class PartitionReservationKey {
/** Cache name. */
private final String cacheName;
@@ -36,7 +36,7 @@ public class MapReservationKey {
* @param cacheName Cache name.
* @param topVer Topology version.
*/
- public MapReservationKey(String cacheName, AffinityTopologyVersion topVer) {
+ public PartitionReservationKey(String cacheName, AffinityTopologyVersion topVer) {
this.cacheName = cacheName;
this.topVer = topVer;
}
@@ -56,7 +56,7 @@ public class MapReservationKey {
if (o == null || getClass() != o.getClass())
return false;
- MapReservationKey other = (MapReservationKey)o;
+ PartitionReservationKey other = (PartitionReservationKey)o;
return F.eq(cacheName, other.cacheName) && F.eq(topVer, other.topVer);
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/PartitionReservationManager.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/PartitionReservationManager.java
new file mode 100644
index 0000000..b720b57
--- /dev/null
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/PartitionReservationManager.java
@@ -0,0 +1,319 @@
+/*
+ * 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.twostep;
+
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cache.PartitionLossPolicy;
+import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.CacheInvalidStateException;
+import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridReservable;
+import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition;
+import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState;
+import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionsReservation;
+import org.apache.ignite.internal.util.typedef.CI1;
+import org.apache.ignite.internal.util.typedef.F;
+import org.jetbrains.annotations.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+import static org.apache.ignite.cache.PartitionLossPolicy.READ_ONLY_SAFE;
+import static org.apache.ignite.cache.PartitionLossPolicy.READ_WRITE_SAFE;
+import static org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion.NONE;
+import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.LOST;
+import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.OWNING;
+
+/**
+ * Class responsible for partition reservation for queries executed on local node.
+ * Prevents partitions from being eveicted from node during query execution.
+ */
+public class PartitionReservationManager {
+ /** Special instance of reservable object for REPLICATED caches. */
+ private static final ReplicatedReservable REPLICATED_RESERVABLE = new ReplicatedReservable();
+
+ /** Kernal context. */
+ private final GridKernalContext ctx;
+
+ /** Reservations. */
+ private final ConcurrentMap<PartitionReservationKey, GridReservable> reservations = new ConcurrentHashMap<>();
+
+ /**
+ * Constructor.
+ *
+ * @param ctx Context.
+ */
+ public PartitionReservationManager(GridKernalContext ctx) {
+ this.ctx = ctx;
+ }
+
+ /**
+ * @param cacheIds Cache IDs.
+ * @param topVer Topology version.
+ * @param explicitParts Explicit partitions list.
+ * @param nodeId Node ID.
+ * @param reqId Request ID.
+ * @return String which is null in case of success or with causeMessage if failed
+ * @throws IgniteCheckedException If failed.
+ */
+ public PartitionReservation reservePartitions(
+ @Nullable List<Integer> cacheIds,
+ AffinityTopologyVersion topVer,
+ final int[] explicitParts,
+ UUID nodeId,
+ long reqId
+ ) throws IgniteCheckedException {
+ assert topVer != null;
+
+ if (F.isEmpty(cacheIds))
+ return new PartitionReservation(Collections.emptyList());
+
+ List<GridReservable> reserved = new ArrayList<>();
+
+ Collection<Integer> partIds;
+
+ if (explicitParts == null)
+ partIds = null;
+ else if (explicitParts.length == 0)
+ partIds = Collections.emptyList();
+ else {
+ partIds = new ArrayList<>(explicitParts.length);
+
+ for (int explicitPart : explicitParts)
+ partIds.add(explicitPart);
+ }
+
+ for (int i = 0; i < cacheIds.size(); i++) {
+ GridCacheContext<?, ?> cctx = ctx.cache().context().cacheContext(cacheIds.get(i));
+
+ // Cache was not found, probably was not deployed yet.
+ if (cctx == null) {
+ return new PartitionReservation(String.format("Failed to reserve partitions for query (cache is not " +
+ "found on local node) [localNodeId=%s, rmtNodeId=%s, reqId=%s, affTopVer=%s, cacheId=%s]",
+ ctx.localNodeId(), nodeId, reqId, topVer, cacheIds.get(i)));
+ }
+
+ if (cctx.isLocal() || !cctx.rebalanceEnabled())
+ continue;
+
+ // For replicated cache topology version does not make sense.
+ final PartitionReservationKey grpKey = new PartitionReservationKey(cctx.name(), cctx.isReplicated() ? null : topVer);
+
+ GridReservable r = reservations.get(grpKey);
+
+ if (explicitParts == null && r != null) { // Try to reserve group partition if any and no explicits.
+ if (r != REPLICATED_RESERVABLE) {
+ if (!r.reserve())
+ return new PartitionReservation(String.format("Failed to reserve partitions for query (group " +
+ "reservation failed) [localNodeId=%s, rmtNodeId=%s, reqId=%s, affTopVer=%s, cacheId=%s, " +
+ "cacheName=%s]",ctx.localNodeId(), nodeId, reqId, topVer, cacheIds.get(i), cctx.name()));
+
+ reserved.add(r);
+ }
+ }
+ else { // Try to reserve partitions one by one.
+ int partsCnt = cctx.affinity().partitions();
+
+ if (cctx.isReplicated()) { // Check all the partitions are in owning state for replicated cache.
+ if (r == null) { // Check only once.
+ for (int p = 0; p < partsCnt; p++) {
+ GridDhtLocalPartition part = partition(cctx, p);
+
+ // We don't need to reserve partitions because they will not be evicted in replicated caches.
+ GridDhtPartitionState partState = part != null ? part.state() : null;
+
+ if (partState != OWNING)
+ return new PartitionReservation(String.format("Failed to reserve partitions for " +
+ "query (partition of REPLICATED cache is not in OWNING state) [" +
+ "localNodeId=%s, rmtNodeId=%s, reqId=%s, affTopVer=%s, cacheId=%s, " +
+ "cacheName=%s, part=%s, partFound=%s, partState=%s]",
+ ctx.localNodeId(),
+ nodeId,
+ reqId,
+ topVer,
+ cacheIds.get(i),
+ cctx.name(),
+ p,
+ (part != null),
+ partState
+ ));
+ }
+
+ // Mark that we checked this replicated cache.
+ reservations.putIfAbsent(grpKey, REPLICATED_RESERVABLE);
+ }
+ }
+ else { // Reserve primary partitions for partitioned cache (if no explicit given).
+ if (explicitParts == null)
+ partIds = cctx.affinity().primaryPartitions(ctx.localNodeId(), topVer);
+
+ int reservedCnt = 0;
+
+ for (int partId : partIds) {
+ GridDhtLocalPartition part = partition(cctx, partId);
+
+ GridDhtPartitionState partState = part != null ? part.state() : null;
+
+ if (partState != OWNING) {
+ if (partState == LOST)
+ ignoreLostPartitionIfPossible(cctx, part);
+ else {
+ return new PartitionReservation(String.format("Failed to reserve partitions " +
+ "for query (partition of PARTITIONED cache is not found or not in OWNING " +
+ "state) [localNodeId=%s, rmtNodeId=%s, reqId=%s, affTopVer=%s, cacheId=%s, " +
+ "cacheName=%s, part=%s, partFound=%s, partState=%s]",
+ ctx.localNodeId(),
+ nodeId,
+ reqId,
+ topVer,
+ cacheIds.get(i),
+ cctx.name(),
+ partId,
+ (part != null),
+ partState
+ ));
+ }
+ }
+
+ if (!part.reserve()) {
+ return new PartitionReservation(String.format("Failed to reserve partitions for query " +
+ "(partition of PARTITIONED cache cannot be reserved) [" +
+ "localNodeId=%s, rmtNodeId=%s, reqId=%s, affTopVer=%s, cacheId=%s, " +
+ "cacheName=%s, part=%s, partFound=%s, partState=%s]",
+ ctx.localNodeId(),
+ nodeId,
+ reqId,
+ topVer,
+ cacheIds.get(i),
+ cctx.name(),
+ partId,
+ true,
+ partState
+ ));
+ }
+
+ reserved.add(part);
+
+ reservedCnt++;
+
+ // Double check that we are still in owning state and partition contents are not cleared.
+ partState = part.state();
+
+ if (partState != OWNING) {
+ if (partState == LOST)
+ ignoreLostPartitionIfPossible(cctx, part);
+ else {
+ return new PartitionReservation(String.format("Failed to reserve partitions for " +
+ "query (partition of PARTITIONED cache is not in OWNING state after " +
+ "reservation) [localNodeId=%s, rmtNodeId=%s, reqId=%s, affTopVer=%s, " +
+ "cacheId=%s, cacheName=%s, part=%s, partState=%s]",
+ ctx.localNodeId(),
+ nodeId,
+ reqId,
+ topVer,
+ cacheIds.get(i),
+ cctx.name(),
+ partId,
+ partState
+ ));
+ }
+ }
+ }
+
+ if (explicitParts == null && reservedCnt > 0) {
+ // We reserved all the primary partitions for cache, attempt to add group reservation.
+ GridDhtPartitionsReservation grp = new GridDhtPartitionsReservation(topVer, cctx, "SQL");
+
+ if (grp.register(reserved.subList(reserved.size() - reservedCnt, reserved.size()))) {
+ if (reservations.putIfAbsent(grpKey, grp) != null)
+ throw new IllegalStateException("Reservation already exists.");
+
+ grp.onPublish(new CI1<GridDhtPartitionsReservation>() {
+ @Override public void apply(GridDhtPartitionsReservation r) {
+ reservations.remove(grpKey, r);
+ }
+ });
+ }
+ }
+ }
+ }
+ }
+
+ return new PartitionReservation(reserved);
+ }
+
+ /**
+ * @param cacheName Cache name.
+ */
+ public void onCacheStop(String cacheName) {
+ // Drop group reservations.
+ for (PartitionReservationKey grpKey : reservations.keySet()) {
+ if (F.eq(grpKey.cacheName(), cacheName))
+ reservations.remove(grpKey);
+ }
+ }
+
+ /**
+ * Decide whether to ignore or proceed with lost partition.
+ *
+ * @param cctx Cache context.
+ * @param part Partition.
+ * @throws IgniteCheckedException If failed.
+ */
+ private static void ignoreLostPartitionIfPossible(GridCacheContext cctx, GridDhtLocalPartition part)
+ throws IgniteCheckedException {
+ PartitionLossPolicy plc = cctx.config().getPartitionLossPolicy();
+
+ if (plc != null) {
+ if (plc == READ_ONLY_SAFE || plc == READ_WRITE_SAFE) {
+ throw new CacheInvalidStateException("Failed to execute query because cache partition has been " +
+ "lost [cacheName=" + cctx.name() + ", part=" + part + ']');
+ }
+ }
+ }
+
+ /**
+ * @param cctx Cache context.
+ * @param p Partition ID.
+ * @return Partition.
+ */
+ private static GridDhtLocalPartition partition(GridCacheContext<?, ?> cctx, int p) {
+ return cctx.topology().localPartition(p, NONE, false);
+ }
+
+ /**
+ * Mapper fake reservation object for replicated caches.
+ */
+ private static class ReplicatedReservable implements GridReservable {
+ /** {@inheritDoc} */
+ @Override public boolean reserve() {
+ throw new IllegalStateException();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void release() {
+ throw new IllegalStateException();
+ }
+ }
+}
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/h2/twostep/RetryCauseMessageSelfTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/h2/twostep/RetryCauseMessageSelfTest.java
index 89894ae..65c37b0 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/h2/twostep/RetryCauseMessageSelfTest.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/h2/twostep/RetryCauseMessageSelfTest.java
@@ -128,13 +128,13 @@ public class RetryCauseMessageSelfTest extends AbstractIndexingCommonTest {
public void testGrpReservationFailureMessage() {
final GridMapQueryExecutor mapQryExec = GridTestUtils.getFieldValue(h2Idx, IgniteH2Indexing.class, "mapQryExec");
- final ConcurrentMap<MapReservationKey, GridReservable> reservations = GridTestUtils.getFieldValue(mapQryExec, GridMapQueryExecutor.class, "reservations");
+ final ConcurrentMap<PartitionReservationKey, GridReservable> reservations = reservations(h2Idx);
GridTestUtils.setFieldValue(h2Idx, IgniteH2Indexing.class, "mapQryExec",
new MockGridMapQueryExecutor(null) {
@Override public void onMessage(UUID nodeId, Object msg) {
if (GridH2QueryRequest.class.isAssignableFrom(msg.getClass())) {
- final MapReservationKey grpKey = new MapReservationKey(ORG, null);
+ final PartitionReservationKey grpKey = new PartitionReservationKey(ORG, null);
reservations.put(grpKey, new GridReservable() {
@@ -276,13 +276,13 @@ public class RetryCauseMessageSelfTest extends AbstractIndexingCommonTest {
public void testNonCollocatedFailureMessage() {
final GridMapQueryExecutor mapQryExec = GridTestUtils.getFieldValue(h2Idx, IgniteH2Indexing.class, "mapQryExec");
- final ConcurrentMap<MapReservationKey, GridReservable> reservations = GridTestUtils.getFieldValue(mapQryExec, GridMapQueryExecutor.class, "reservations");
+ final ConcurrentMap<PartitionReservationKey, GridReservable> reservations = reservations(h2Idx);
GridTestUtils.setFieldValue(h2Idx, IgniteH2Indexing.class, "mapQryExec",
new MockGridMapQueryExecutor(null) {
@Override public void onMessage(UUID nodeId, Object msg) {
if (GridH2QueryRequest.class.isAssignableFrom(msg.getClass())) {
- final MapReservationKey grpKey = new MapReservationKey(ORG, null);
+ final PartitionReservationKey grpKey = new PartitionReservationKey(ORG, null);
reservations.put(grpKey, new GridReservable() {
@@ -294,6 +294,7 @@ public class RetryCauseMessageSelfTest extends AbstractIndexingCommonTest {
}
});
}
+
startedExecutor.onMessage(nodeId, msg);
}
@@ -363,20 +364,30 @@ public class RetryCauseMessageSelfTest extends AbstractIndexingCommonTest {
stopAllGrids();
}
+ /**
+ * @param h2Idx Indexing.
+ * @return Current reservations.
+ */
+ private static ConcurrentMap<PartitionReservationKey, GridReservable> reservations(IgniteH2Indexing h2Idx) {
+ PartitionReservationManager partReservationMgr = h2Idx.partitionReservationManager();
+
+ return GridTestUtils.getFieldValue(partReservationMgr, PartitionReservationManager.class, "reservations");
+ }
/**
* Wrapper around @{GridMapQueryExecutor}
*/
private abstract static class MockGridMapQueryExecutor extends GridMapQueryExecutor {
+ /** Wrapped executor */
+ GridMapQueryExecutor startedExecutor;
/**
- * Wrapped executor
+ * @param startedExecutor Started executor.
+ * @return Mocked map query executor.
*/
- GridMapQueryExecutor startedExecutor;
+ MockGridMapQueryExecutor insertRealExecutor(GridMapQueryExecutor startedExecutor) {
+ this.startedExecutor = startedExecutor;
- /** */
- MockGridMapQueryExecutor insertRealExecutor(GridMapQueryExecutor realExecutor) {
- this.startedExecutor = realExecutor;
return this;
}
@@ -403,11 +414,6 @@ public class RetryCauseMessageSelfTest extends AbstractIndexingCommonTest {
}
/** {@inheritDoc} */
- @Override public void onCacheStop(String cacheName) {
- startedExecutor.onCacheStop(cacheName);
- }
-
- /** {@inheritDoc} */
@Override public void stopAndUnregisterCurrentLazyWorker() {
startedExecutor.stopAndUnregisterCurrentLazyWorker();
}