ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From se...@apache.org
Subject [24/50] incubator-ignite git commit: Merge branch 'ignite-sprint-5' of https://git-wip-us.apache.org/repos/asf/incubator-ignite into ignite-484-1
Date Thu, 18 Jun 2015 06:54:31 GMT
Merge branch 'ignite-sprint-5' of https://git-wip-us.apache.org/repos/asf/incubator-ignite
into ignite-484-1

# Conflicts:
#	modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java
#	modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java


Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/51bf4b15
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/51bf4b15
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/51bf4b15

Branch: refs/heads/ignite-sprint-6
Commit: 51bf4b1565a263a919635ab34f09f8673d8e42c9
Parents: b5db559 89a4f7c
Author: S.Vladykin <svladykin@gridgain.com>
Authored: Thu Jun 11 10:00:07 2015 +0300
Committer: S.Vladykin <svladykin@gridgain.com>
Committed: Thu Jun 11 10:00:07 2015 +0300

----------------------------------------------------------------------
 DEVNOTES.txt                                    |   6 +
 assembly/dependencies-fabric.xml                |   1 +
 examples/pom.xml                                |  34 ++
 modules/core/pom.xml                            |   1 -
 .../apache/ignite/cache/query/ScanQuery.java    |  45 +-
 .../configuration/CacheConfiguration.java       |   1 -
 .../processors/cache/GridCacheAdapter.java      |  15 +-
 .../processors/cache/GridCacheProcessor.java    |  30 +-
 .../processors/cache/GridCacheSwapManager.java  |  55 ++-
 .../processors/cache/IgniteCacheProxy.java      |  11 +-
 .../processors/cache/QueryCursorImpl.java       |  23 +-
 .../distributed/dht/GridDhtLocalPartition.java  |   7 +
 .../processors/cache/query/CacheQuery.java      |   2 +-
 .../query/GridCacheDistributedQueryManager.java |   3 +
 .../cache/query/GridCacheQueryAdapter.java      | 147 ++++++-
 .../cache/query/GridCacheQueryManager.java      | 209 ++++++----
 .../cache/query/GridCacheQueryRequest.java      |  47 ++-
 .../processors/cache/query/QueryCursorEx.java   |   8 +
 .../datastructures/GridCacheSetImpl.java        |   4 +-
 .../processors/query/GridQueryIndexing.java     |   4 +-
 .../processors/query/GridQueryProcessor.java    |  18 +-
 .../service/GridServiceProcessor.java           |   2 +-
 .../ignite/internal/util/GridJavaProcess.java   |   2 +-
 .../ignite/internal/util/IgniteUtils.java       |   4 +-
 .../shmem/IpcSharedMemoryClientEndpoint.java    |   2 +-
 .../ipc/shmem/IpcSharedMemoryNativeLoader.java  | 151 ++++++-
 .../shmem/IpcSharedMemoryServerEndpoint.java    |   2 +-
 .../util/nio/GridShmemCommunicationClient.java  | 146 +++++++
 .../communication/tcp/TcpCommunicationSpi.java  | 415 ++++++++++++++++++-
 .../tcp/TcpCommunicationSpiMBean.java           |   8 +
 .../cache/GridCacheAbstractFullApiSelfTest.java |  15 +
 .../cache/IgniteDynamicCacheStartSelfTest.java  |  19 +
 .../IgniteCacheMessageRecoveryAbstractTest.java |   1 +
 ...achePartitionedPreloadLifecycleSelfTest.java |   2 +-
 ...CacheReplicatedPreloadLifecycleSelfTest.java |   6 +-
 .../GridCacheSwapScanQueryAbstractSelfTest.java | 112 +++--
 .../ipc/shmem/IgfsSharedMemoryTestServer.java   |   2 +
 .../IpcSharedMemoryCrashDetectionSelfTest.java  |   2 +-
 .../ipc/shmem/IpcSharedMemorySpaceSelfTest.java |   2 +-
 .../ipc/shmem/IpcSharedMemoryUtilsSelfTest.java |   2 +-
 .../LoadWithCorruptedLibFileTestRunner.java     |   2 +-
 .../IpcSharedMemoryBenchmarkReader.java         |   2 +-
 .../IpcSharedMemoryBenchmarkWriter.java         |   2 +-
 .../communication/GridIoManagerBenchmark0.java  |   1 +
 .../spi/GridTcpSpiForwardingSelfTest.java       |   1 +
 .../GridTcpCommunicationSpiAbstractTest.java    |  13 +
 ...mmunicationSpiConcurrentConnectSelfTest.java |   4 +-
 ...cpCommunicationSpiMultithreadedSelfTest.java |  21 +-
 ...pCommunicationSpiMultithreadedShmemTest.java |  28 ++
 ...dTcpCommunicationSpiRecoveryAckSelfTest.java |   1 +
 ...GridTcpCommunicationSpiRecoverySelfTest.java |   1 +
 .../GridTcpCommunicationSpiShmemSelfTest.java   |  38 ++
 .../tcp/GridTcpCommunicationSpiTcpSelfTest.java |   7 +
 .../IgniteSpiCommunicationSelfTestSuite.java    |   2 +
 modules/hadoop/pom.xml                          |   1 +
 .../HadoopIgfs20FileSystemAbstractSelfTest.java |  13 +
 ...oopSecondaryFileSystemConfigurationTest.java |  14 +
 ...IgniteHadoopFileSystemHandshakeSelfTest.java |   7 +
 .../IgniteHadoopFileSystemIpcCacheSelfTest.java |   7 +
 .../hadoop/HadoopAbstractSelfTest.java          |   7 +
 .../processors/query/h2/IgniteH2Indexing.java   |  44 +-
 .../h2/twostep/GridReduceQueryExecutor.java     |  23 +-
 ...CacheScanPartitionQueryFallbackSelfTest.java | 408 ++++++++++++++++++
 .../cache/GridCacheCrossCacheQuerySelfTest.java |  12 +-
 .../cache/IgniteCacheAbstractQuerySelfTest.java |  77 +++-
 .../IgniteCacheQuerySelfTestSuite.java          |   2 +
 modules/scalar-2.10/README.txt                  |   4 +
 modules/scalar-2.10/licenses/apache-2.0.txt     | 202 +++++++++
 .../scalar-2.10/licenses/scala-bsd-license.txt  |  18 +
 modules/scalar-2.10/pom.xml                     | 197 +++++++++
 modules/spark-2.10/README.txt                   |   4 +
 modules/spark-2.10/licenses/apache-2.0.txt      | 202 +++++++++
 .../spark-2.10/licenses/scala-bsd-license.txt   |  18 +
 modules/spark-2.10/pom.xml                      | 120 ++++++
 modules/spark/README.txt                        |   8 +
 modules/spark/licenses/apache-2.0.txt           | 202 +++++++++
 modules/spark/licenses/scala-bsd-license.txt    |  18 +
 modules/spark/pom.xml                           | 114 +++++
 .../org/apache/ignite/spark/IgniteContext.scala | 119 ++++++
 .../org/apache/ignite/spark/IgniteRDD.scala     | 244 +++++++++++
 .../apache/ignite/spark/JavaIgniteContext.scala |  63 +++
 .../org/apache/ignite/spark/JavaIgniteRDD.scala |  99 +++++
 .../ignite/spark/impl/IgniteAbstractRDD.scala   |  39 ++
 .../ignite/spark/impl/IgnitePartition.scala     |  24 ++
 .../ignite/spark/impl/IgniteQueryIterator.scala |  27 ++
 .../apache/ignite/spark/impl/IgniteSqlRDD.scala |  41 ++
 .../spark/impl/JavaIgniteAbstractRDD.scala      |  34 ++
 .../ignite/spark/JavaIgniteRDDSelfTest.java     | 298 +++++++++++++
 .../scala/org/apache/ignite/spark/Entity.scala  |  28 ++
 .../org/apache/ignite/spark/IgniteRddSpec.scala | 231 +++++++++++
 modules/visor-console-2.10/README.txt           |   4 +
 modules/visor-console-2.10/pom.xml              | 174 ++++++++
 parent/pom.xml                                  |   4 +
 pom.xml                                         |  20 +-
 94 files changed, 4595 insertions(+), 265 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/51bf4b15/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/51bf4b15/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/51bf4b15/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryIndexing.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/51bf4b15/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java
index 49d2401,e187713..282875d
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java
@@@ -706,8 -718,7 +709,8 @@@ public class GridQueryProcessor extend
                      String sql = qry.getSql();
                      Object[] args = qry.getArgs();
  
-                     GridQueryFieldsResult res = idx.queryFields(space, sql, F.asList(args),
 -                    final GridQueryFieldsResult res = idx.queryFields(space, sql, F.asList(args),
idx.backupFilter());
++                    final GridQueryFieldsResult res = idx.queryFields(space, sql, F.asList(args),
 +                        idx.backupFilter(null, null, null));
  
                      sendQueryExecutedEvent(sql, args);
  

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/51bf4b15/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/51bf4b15/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java
----------------------------------------------------------------------
diff --cc modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java
index 03da6d3,11054b7..3d2ae46
--- 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
@@@ -397,353 -269,109 +397,338 @@@ public class GridReduceQueryExecutor 
       * @param qry Query.
       * @return Cursor.
       */
-     public QueryCursor<List<?>> query(GridCacheContext<?,?> cctx, GridCacheTwoStepQuery
qry) {
 -    public Iterator<List<?>> query(GridCacheContext<?,?> cctx, GridCacheTwoStepQuery
qry, boolean keepPortable) {
 -        long qryReqId = reqIdGen.incrementAndGet();
++    public QueryCursor<List<?>> query(GridCacheContext<?,?> cctx, GridCacheTwoStepQuery
qry, boolean keepPortable) {
 +        for (;;) {
 +            long qryReqId = reqIdGen.incrementAndGet();
  
 -        QueryRun r = new QueryRun();
 +            QueryRun r = new QueryRun();
  
 -        r.pageSize = qry.pageSize() <= 0 ? GridCacheTwoStepQuery.DFLT_PAGE_SIZE : qry.pageSize();
 +            r.pageSize = qry.pageSize() <= 0 ? GridCacheTwoStepQuery.DFLT_PAGE_SIZE :
qry.pageSize();
  
 -        r.tbls = new ArrayList<>(qry.mapQueries().size());
 +            r.tbls = new ArrayList<>(qry.mapQueries().size());
  
 -        String space = cctx.name();
 +            String space = cctx.name();
  
 -        r.conn = (JdbcConnection)h2.connectionForSpace(space);
 +            r.conn = (JdbcConnection)h2.connectionForSpace(space);
  
 -        // TODO    Add topology version.
 -        ClusterGroup dataNodes = ctx.grid().cluster().forDataNodes(space);
 +            AffinityTopologyVersion topVer = h2.readyTopologyVersion();
  
 -        if (cctx.isReplicated() || qry.explain()) {
 -            assert qry.explain() || dataNodes.node(ctx.localNodeId()) == null : "We must
be on a client node.";
 +            List<String> extraSpaces = extraSpaces(space, qry.spaces());
  
 -            // Select random data node to run query on a replicated data or get EXPLAIN
PLAN from a single node.
 -            dataNodes = dataNodes.forRandom();
 -        }
 +            Collection<ClusterNode> nodes;
  
 -        final Collection<ClusterNode> nodes = dataNodes.nodes();
 +            // Explicit partition mapping for unstable topology.
 +            Map<ClusterNode, IntArray> partsMap = null;
  
 -        for (GridCacheSqlQuery mapQry : qry.mapQueries()) {
 -            GridMergeTable tbl;
 +            if (isPreloadingActive(cctx, extraSpaces)) {
 +                if (cctx.isReplicated())
 +                    nodes = replicatedDataNodes(cctx, extraSpaces);
 +                else {
 +                    partsMap = partitionLocations(cctx, extraSpaces);
  
 -            try {
 -                tbl = createFunctionTable(r.conn, mapQry, qry.explain()); // createTable(r.conn,
mapQry); TODO
 +                    nodes = partsMap == null ? null : partsMap.keySet();
 +                }
              }
 -            catch (IgniteCheckedException e) {
 -                throw new IgniteException(e);
 +            else
 +                nodes = stableDataNodes(topVer, cctx, extraSpaces);
 +
 +            if (nodes == null)
 +                continue; // Retry.
 +
 +            assert !nodes.isEmpty();
 +
 +            if (cctx.isReplicated() || qry.explain()) {
 +                assert qry.explain() || !nodes.contains(ctx.cluster().get().localNode())
: "We must be on a client node.";
 +
 +                // Select random data node to run query on a replicated data or get EXPLAIN
PLAN from a single node.
 +                nodes = Collections.singleton(F.rand(nodes));
 +            }
 +
 +            for (GridCacheSqlQuery mapQry : qry.mapQueries()) {
 +                GridMergeTable tbl;
 +
 +                try {
 +                    tbl = createFunctionTable(r.conn, mapQry, qry.explain()); // createTable(r.conn,
mapQry); TODO
 +                }
 +                catch (IgniteCheckedException e) {
 +                    throw new IgniteException(e);
 +                }
 +
 +                GridMergeIndex idx = tbl.getScanIndex(null);
 +
 +                for (ClusterNode node : nodes)
 +                    idx.addSource(node.id());
 +
 +                r.tbls.add(tbl);
 +
 +                curFunTbl.set(tbl);
              }
  
 -            GridMergeIndex idx = tbl.getScanIndex(null);
 +            r.latch = new CountDownLatch(r.tbls.size() * nodes.size());
 +
 +            runs.put(qryReqId, r);
 +
 +            try {
 +                Collection<GridCacheSqlQuery> mapQrys = qry.mapQueries();
 +
 +                if (qry.explain()) {
 +                    mapQrys = new ArrayList<>(qry.mapQueries().size());
 +
 +                    for (GridCacheSqlQuery mapQry : qry.mapQueries())
 +                        mapQrys.add(new GridCacheSqlQuery(mapQry.alias(), "EXPLAIN " + mapQry.query(),
mapQry.parameters()));
 +                }
 +
 +                if (nodes.size() != 1 || !F.first(nodes).isLocal()) { // Marshall params
for remotes.
 +                    Marshaller m = ctx.config().getMarshaller();
 +
 +                    for (GridCacheSqlQuery mapQry : mapQrys)
 +                        mapQry.marshallParams(m);
 +                }
 +
 +                boolean retry = false;
 +
 +                if (send(nodes,
 +                    new GridQueryRequest(qryReqId, r.pageSize, space, mapQrys, topVer, extraSpaces,
null), partsMap)) {
 +                    U.await(r.latch);
 +
 +                    Object state = r.state.get();
 +
 +                    if (state != null) {
 +                        if (state instanceof CacheException)
 +                            throw new CacheException("Failed to run map query remotely.",
(CacheException)state);
 +
 +                        if (state instanceof AffinityTopologyVersion) {
 +                            retry = true;
 +
 +                            // If remote node asks us to retry then we have outdated full
partition map.
 +                            // TODO is this correct way to wait for a new map??
 +                            h2.awaitForReadyTopologyVersion((AffinityTopologyVersion)state);
 +                        }
 +                    }
 +                }
 +                else // Send failed.
 +                    retry = true;
 +
 +                ResultSet res = null;
 +
 +                if (!retry) {
 +                    if (qry.explain())
 +                        return explainPlan(r.conn, space, qry);
 +
 +                    GridCacheSqlQuery rdc = qry.reduceQuery();
 +
 +                    res = h2.executeSqlQueryWithTimer(space, r.conn, rdc.query(), F.asList(rdc.parameters()));
 +                }
 +
 +                for (GridMergeTable tbl : r.tbls) {
 +                    if (!tbl.getScanIndex(null).fetchedAll()) // We have to explicitly cancel
queries on remote nodes.
 +                        send(nodes, new GridQueryCancelRequest(qryReqId), null);
 +
 +//                dropTable(r.conn, tbl.getName()); TODO
 +                }
  
 -            for (ClusterNode node : nodes)
 -                idx.addSource(node.id());
 +                if (retry) {
 +                    if (Thread.currentThread().isInterrupted())
 +                        throw new IgniteInterruptedCheckedException("Query was interrupted.");
  
 -            r.tbls.add(tbl);
 +                    continue;
 +                }
 +
 +                return new QueryCursorImpl<>(new GridQueryCacheObjectsIterator(new
Iter(res), cctx, cctx.keepPortable()));
 +            }
 +            catch (IgniteCheckedException | RuntimeException e) {
 +                U.closeQuiet(r.conn);
  
 -            curFunTbl.set(tbl);
 +                if (e instanceof CacheException)
 +                    throw (CacheException)e;
 +
 +                throw new CacheException("Failed to run reduce query locally.", e);
 +            }
 +            finally {
 +                if (!runs.remove(qryReqId, r))
 +                    U.warn(log, "Query run was already removed: " + qryReqId);
 +
 +                curFunTbl.remove();
 +            }
          }
 +    }
 +
 +    /**
 +     * Calculates data nodes for replicated caches on unstable topology.
 +     *
 +     * @param cctx Cache context for main space.
 +     * @param extraSpaces Extra spaces.
 +     * @return Collection of all data nodes owning all the caches or {@code null} for retry.
 +     */
 +    private Collection<ClusterNode> replicatedDataNodes(final GridCacheContext<?,?>
cctx, List<String> extraSpaces) {
 +        assert cctx.isReplicated() : cctx.name() + " must be replicated";
  
 -        r.latch = new CountDownLatch(r.tbls.size() * nodes.size());
 +        Set<ClusterNode> nodes = owningReplicatedDataNodes(cctx);
  
 -        runs.put(qryReqId, r);
 +        if (!F.isEmpty(extraSpaces)) {
 +            for (String extraSpace : extraSpaces) {
 +                GridCacheContext<?,?> extraCctx = cacheContext(extraSpace);
  
 -        try {
 -            Collection<GridCacheSqlQuery> mapQrys = qry.mapQueries();
 +                if (extraCctx.isLocal())
 +                    continue;
  
 -            if (qry.explain()) {
 -                mapQrys = new ArrayList<>(qry.mapQueries().size());
 +                if (!extraCctx.isReplicated())
 +                    throw new CacheException("Queries running on replicated cache should
not contain JOINs " +
 +                        "with partitioned tables.");
  
 -                for (GridCacheSqlQuery mapQry : qry.mapQueries())
 -                    mapQrys.add(new GridCacheSqlQuery(mapQry.alias(), "EXPLAIN " + mapQry.query(),
mapQry.parameters()));
 +                nodes.retainAll(owningReplicatedDataNodes(extraCctx));
 +
 +                if (nodes.isEmpty())
 +                    return null; // Retry.
              }
 +        }
  
 -            if (nodes.size() != 1 || !F.first(nodes).isLocal()) { // Marshall params for
remotes.
 -                Marshaller m = ctx.config().getMarshaller();
 +        return nodes;
 +    }
  
 -                for (GridCacheSqlQuery mapQry : mapQrys)
 -                    mapQry.marshallParams(m);
 +    /**
 +     * Collects all the nodes owning all the partitions for the given replicated cache.
 +     *
 +     * @param cctx Cache context.
 +     * @return Owning nodes.
 +     */
 +    private Set<ClusterNode> owningReplicatedDataNodes(GridCacheContext<?,?>
cctx) {
 +        assert cctx.isReplicated() : cctx.name() + " must be replicated";
 +
 +        String space = cctx.name();
 +
 +        Set<ClusterNode> dataNodes = new HashSet<>(ctx.discovery().cacheAffinityNodes(space,
NONE));
 +
 +        if (dataNodes.isEmpty())
 +            throw new CacheException("No data nodes found for cache '" + space + "'");
 +
 +        // Find all the nodes owning all the partitions for replicated cache.
 +        for (int p = 0, extraParts = cctx.affinity().partitions(); p < extraParts; p++)
{
 +            List<ClusterNode> owners = cctx.topology().owners(p);
 +
 +            if (owners.isEmpty())
 +                throw new CacheException("No data nodes found for cache '" + space +
 +                    "' for partition " + p);
 +
 +            dataNodes.retainAll(owners);
 +
 +            if (dataNodes.isEmpty())
 +                throw new CacheException("No data nodes found for cache '" + space +
 +                    "' owning all the partitions.");
 +        }
 +
 +        return dataNodes;
 +    }
 +
 +    /**
 +     * Calculates partition mapping for partitioned cache on unstable topology.
 +     *
 +     * @param cctx Cache context for main space.
 +     * @param extraSpaces Extra spaces.
 +     * @return Partition mapping or {@code null} if we can't calculate it due to repartitioning
and we need to retry.
 +     */
 +    @SuppressWarnings("unchecked")
 +    private Map<ClusterNode, IntArray> partitionLocations(final GridCacheContext<?,?>
cctx, List<String> extraSpaces) {
 +        assert !cctx.isReplicated() && !cctx.isLocal() : cctx.name() + " must be
partitioned";
 +
 +        final int partsCnt = cctx.affinity().partitions();
 +
 +        if (extraSpaces != null) { // Check correct number of partitions for partitioned
caches.
 +            for (String extraSpace : extraSpaces) {
 +                GridCacheContext<?,?> extraCctx = cacheContext(extraSpace);
 +
 +                if (extraCctx.isReplicated() || extraCctx.isLocal())
 +                    continue;
 +
 +                int parts = extraCctx.affinity().partitions();
 +
 +                if (parts != partsCnt)
 +                    throw new CacheException("Number of partitions must be the same for
correct collocation in " +
 +                        "caches " + cctx.name() + " and " + extraSpace + ".");
              }
 +        }
  
 -            send(nodes, new GridQueryRequest(qryReqId, r.pageSize, space, mapQrys));
 +        Set<ClusterNode>[] partLocs = new Set[partsCnt];
  
 -            r.latch.await();
 +        // Fill partition locations for main cache.
 +        for (int p = 0, parts =  cctx.affinity().partitions(); p < parts; p++) {
 +            List<ClusterNode> owners = cctx.topology().owners(p);
  
 -            if (r.rmtErr != null)
 -                throw new CacheException("Failed to run map query remotely.", r.rmtErr);
 +            if (F.isEmpty(owners))
 +                throw new CacheException("No data nodes found for cache '" + cctx.name()
+ "' for partition " + p);
  
 -            if (qry.explain())
 -                return explainPlan(r.conn, space, qry);
 +            partLocs[p] = new HashSet<>(owners);
 +        }
  
 -            GridCacheSqlQuery rdc = qry.reduceQuery();
 +        if (extraSpaces != null) {
 +            // Find owner intersections for each participating partitioned cache partition.
 +            // We need this for logical collocation between different partitioned caches
with the same affinity.
 +            for (String extraSpace : extraSpaces) {
 +                GridCacheContext<?,?> extraCctx = cacheContext(extraSpace);
  
 -            final ResultSet res = h2.executeSqlQueryWithTimer(space, r.conn, rdc.query(),
F.asList(rdc.parameters()));
 +                if (extraCctx.isReplicated() || extraCctx.isLocal())
 +                    continue;
  
 -            for (GridMergeTable tbl : r.tbls) {
 -                if (!tbl.getScanIndex(null).fetchedAll()) // We have to explicitly cancel
queries on remote nodes.
 -                    send(nodes, new GridQueryCancelRequest(qryReqId));
 +                for (int p = 0, parts =  extraCctx.affinity().partitions(); p < parts;
p++) {
 +                    List<ClusterNode> owners = extraCctx.topology().owners(p);
  
 -//                dropTable(r.conn, tbl.getName()); TODO
 +                    if (F.isEmpty(owners))
 +                        throw new CacheException("No data nodes found for cache '" + extraSpace
+
 +                            "' for partition " + p);
 +
 +                    if (partLocs[p] == null)
 +                        partLocs[p] = new HashSet<>(owners);
 +                    else {
 +                        partLocs[p].retainAll(owners); // Intersection of owners.
 +
 +                        if (partLocs[p].isEmpty())
 +                            return null; // Intersection is empty -> retry.
 +                    }
 +                }
              }
  
-             // Filter nodes where not all the replicated caches loaded.
-             for (String extraSpace : extraSpaces) {
-                 GridCacheContext<?,?> extraCctx = cacheContext(extraSpace);
- 
-                 if (!extraCctx.isReplicated())
-                     continue;
- 
-                 Set<ClusterNode> dataNodes = owningReplicatedDataNodes(extraCctx);
- 
-                 for (Set<ClusterNode> partLoc : partLocs) {
-                     partLoc.retainAll(dataNodes);
- 
-                     if (partLoc.isEmpty())
-                         return null; // Retry.
-                 }
-             }
 -            return new GridQueryCacheObjectsIterator(new Iter(res), cctx, keepPortable);
++            return new QueryCursorImpl<>(new GridQueryCacheObjectsIterator(new Iter(res),
cctx, cctx.keepPortable()));
          }
 -        catch (IgniteCheckedException | InterruptedException | RuntimeException e) {
 -            U.closeQuiet(r.conn);
  
 -            if (e instanceof CacheException)
 -                throw (CacheException)e;
 +        // Collect the final partitions mapping.
 +        Map<ClusterNode, IntArray> res = new HashMap<>();
  
 -            throw new CacheException("Failed to run reduce query locally.", e);
 +        // Here partitions in all IntArray's will be sorted in ascending order, this is
important.
 +        for (int p = 0; p < partLocs.length; p++) {
 +            Set<ClusterNode> pl = partLocs[p];
 +
 +            assert !F.isEmpty(pl) : pl;
 +
 +            ClusterNode n = pl.size() == 1 ? F.first(pl) : F.rand(pl);
 +
 +            IntArray parts = res.get(n);
 +
 +            if (parts == null)
 +                res.put(n, parts = new IntArray());
 +
 +            parts.add(p);
          }
 -        finally {
 -            if (!runs.remove(qryReqId, r))
 -                U.warn(log, "Query run was already removed: " + qryReqId);
  
 -            curFunTbl.remove();
 +        return res;
 +    }
 +
 +    /**
 +     * @param mainSpace Main space.
 +     * @param allSpaces All spaces.
 +     * @return List of all extra spaces or {@code null} if none.
 +     */
 +    private List<String> extraSpaces(String mainSpace, Set<String> allSpaces)
{
 +        if (F.isEmpty(allSpaces) || (allSpaces.size() == 1 && allSpaces.contains(mainSpace)))
 +            return null;
 +
 +        ArrayList<String> res = new ArrayList<>(allSpaces.size());
 +
 +        for (String space : allSpaces) {
 +            if (!F.eq(space, mainSpace))
 +                res.add(space);
          }
 +
 +        return res;
      }
  
      /**

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/51bf4b15/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheCrossCacheQuerySelfTest.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/51bf4b15/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite.java
----------------------------------------------------------------------


Mime
View raw message