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 2017/08/14 10:30:18 UTC
[08/50] ignite git commit: IGNITE-6019 Merge indexes iterator
IGNITE-6019 Merge indexes iterator
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/12a479fe
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/12a479fe
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/12a479fe
Branch: refs/heads/ignite-5991-6019
Commit: 12a479fe7d1973bdba5dcdb488d3ff109667d107
Parents: 3919d80
Author: Alexander Paschenko <al...@gmail.com>
Authored: Thu Aug 10 15:48:38 2017 +0300
Committer: Alexander Paschenko <al...@gmail.com>
Committed: Thu Aug 10 15:48:38 2017 +0300
----------------------------------------------------------------------
.../h2/twostep/GridMergeIndexesIterator.java | 133 +++++++++++++++++++
.../h2/twostep/GridReduceQueryExecutor.java | 62 ++++-----
2 files changed, 158 insertions(+), 37 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/12a479fe/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMergeIndexesIterator.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMergeIndexesIterator.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMergeIndexesIterator.java
new file mode 100644
index 0000000..9d18269
--- /dev/null
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMergeIndexesIterator.java
@@ -0,0 +1,133 @@
+/*
+ * 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 java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.util.GridCloseableIteratorAdapterEx;
+import org.apache.ignite.lang.IgniteOutClosure;
+import org.h2.index.Cursor;
+import org.h2.result.Row;
+
+/**
+ * Iterator that transparently and sequentially traverses a bunch of {@link GridMergeIndex}es.
+ */
+class GridMergeIndexesIterator extends GridCloseableIteratorAdapterEx<List<?>> {
+ /** Iterator over indexes. */
+ private final Iterator<GridMergeIndex> idxs;
+
+ /** Current cursor. */
+ private Cursor cur;
+
+ /** Next row to return. */
+ private List<Object> next;
+
+ /** Closure to run when this iterator is closed, or when all indexes have been traversed. */
+ private final IgniteOutClosure<?> clo;
+
+ /** Cleanup flag - {@code null} iff {@link #clo} is null. */
+ private final AtomicBoolean cleanupDone;
+
+ /**
+ * Constructor.
+ * @param idxs Indexes to iterate over.
+ * @param clo Closure to run when this iterator is closed, or when all indexes have been traversed.
+ */
+ GridMergeIndexesIterator(Iterable<GridMergeIndex> idxs, IgniteOutClosure<?> clo) {
+ this.idxs = idxs.iterator();
+ this.clo = clo;
+
+ cleanupDone = (clo != null ? new AtomicBoolean() : null);
+
+ goNext();
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean onHasNext() throws IgniteCheckedException {
+ return next != null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public List<?> onNext() throws IgniteCheckedException {
+ final List<?> res = next;
+
+ if (res == null)
+ throw new NoSuchElementException();
+
+ next = null;
+
+ try {
+ goNext();
+ }
+ catch (Throwable e) {
+ doCleanup();
+
+ throw new IgniteCheckedException(e);
+ }
+
+ return res;
+ }
+
+ /**
+ * Retrieve next row.
+ */
+ private void goNext() {
+ next = null;
+
+ boolean gotNext = false;
+
+ while (cur == null || !(gotNext = cur.next())) {
+ if (idxs.hasNext())
+ cur = idxs.next().findInStream(null, null);
+ else {
+ // We're out of records, let's clean up.
+ doCleanup();
+
+ break;
+ }
+ }
+
+ if (gotNext) {
+ Row row = cur.get();
+
+ int cols = row.getColumnCount();
+
+ next = new ArrayList<>(cols);
+
+ for (int c = 0; c < cols; c++)
+ next.add(row.getValue(c).getObject());
+ }
+ }
+
+ /** Run {@link #clo} if it's present and hasn't been invoked yet. */
+ private void doCleanup() {
+ if (clo != null && cleanupDone.compareAndSet(false, true))
+ clo.apply();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void onClose() throws IgniteCheckedException {
+ doCleanup();
+
+ super.onClose();
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/12a479fe/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java
----------------------------------------------------------------------
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 85a7e0b..7ca91a5 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
@@ -84,13 +84,12 @@ 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.IgniteOutClosure;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.h2.command.ddl.CreateTableData;
import org.h2.engine.Session;
-import org.h2.index.Cursor;
import org.h2.index.Index;
import org.h2.jdbc.JdbcConnection;
-import org.h2.result.Row;
import org.h2.table.Column;
import org.h2.util.IntArray;
import org.h2.value.Value;
@@ -511,7 +510,7 @@ public class GridReduceQueryExecutor {
*/
public Iterator<List<?>> query(
String schemaName,
- GridCacheTwoStepQuery qry,
+ final GridCacheTwoStepQuery qry,
boolean keepBinary,
boolean enforceJoinOrder,
int timeoutMillis,
@@ -617,11 +616,26 @@ public class GridReduceQueryExecutor {
final boolean skipMergeTbl = !qry.explain() && qry.skipMergeTable();
- final int segmentsPerIndex = qry.explain() || isReplicatedOnly ? 1 :
+ final int segmentsPerIdx = qry.explain() || isReplicatedOnly ? 1 :
findFirstPartitioned(cacheIds).config().getQueryParallelism();
int replicatedQrysCnt = 0;
+ final Collection<ClusterNode> finalNodes = nodes;
+
+ // This will be executed either at the end of lazy iteration or after merge table is filled.
+ final IgniteOutClosure<Void> clo = new IgniteOutClosure<Void>() {
+ @Override public Void apply() {
+ // Make sure any activity related to current attempt is cancelled.
+ cancelRemoteQueriesIfNeeded(finalNodes, r, qryReqId, qry.distributedJoins());
+
+ if (!runs.remove(qryReqId, r))
+ U.warn(log, "Query run was already removed: " + qryReqId);
+
+ return null;
+ }
+ };
+
for (GridCacheSqlQuery mapQry : qry.mapQueries()) {
GridMergeIndex idx;
@@ -653,7 +667,7 @@ public class GridReduceQueryExecutor {
idx.setSources(singletonList(node), 1); // Replicated tables can have only 1 segment.
}
else
- idx.setSources(nodes, segmentsPerIndex);
+ idx.setSources(nodes, segmentsPerIdx);
idx.setPageSize(r.pageSize());
@@ -661,7 +675,7 @@ public class GridReduceQueryExecutor {
}
r.latch(new CountDownLatch(isReplicatedOnly ? 1 :
- (r.indexes().size() - replicatedQrysCnt) * nodes.size() * segmentsPerIndex + replicatedQrysCnt));
+ (r.indexes().size() - replicatedQrysCnt) * nodes.size() * segmentsPerIdx + replicatedQrysCnt));
runs.put(qryReqId, r);
@@ -686,8 +700,6 @@ public class GridReduceQueryExecutor {
final boolean distributedJoins = qry.distributedJoins();
- final Collection<ClusterNode> finalNodes = nodes;
-
cancel.set(new Runnable() {
@Override public void run() {
send(finalNodes, new GridQueryCancelRequest(qryReqId), null, false);
@@ -756,29 +768,8 @@ public class GridReduceQueryExecutor {
Iterator<List<?>> resIter = null;
if (!retry) {
- if (skipMergeTbl) {
- List<List<?>> res = new ArrayList<>();
-
- // Simple UNION ALL can have multiple indexes.
- for (GridMergeIndex idx : r.indexes()) {
- Cursor cur = idx.findInStream(null, null);
-
- while (cur.next()) {
- Row row = cur.get();
-
- int cols = row.getColumnCount();
-
- List<Object> resRow = new ArrayList<>(cols);
-
- for (int c = 0; c < cols; c++)
- resRow.add(row.getValue(c).getObject());
-
- res.add(resRow);
- }
- }
-
- resIter = res.iterator();
- }
+ if (skipMergeTbl)
+ resIter = new GridMergeIndexesIterator(r.indexes(), clo);
else {
cancel.checkCancelled();
@@ -842,13 +833,10 @@ public class GridReduceQueryExecutor {
throw new CacheException("Failed to run reduce query locally.", cause);
}
finally {
- // Make sure any activity related to current attempt is cancelled.
- cancelRemoteQueriesIfNeeded(nodes, r, qryReqId, qry.distributedJoins());
-
- if (!runs.remove(qryReqId, r))
- U.warn(log, "Query run was already removed: " + qryReqId);
-
+ // If we have fetched all data to merge table, then let's free all resources here.
if (!skipMergeTbl) {
+ clo.apply();
+
for (int i = 0, mapQrys = qry.mapQueries().size(); i < mapQrys; i++)
fakeTable(null, i).innerTable(null); // Drop all merge tables.
}