ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [01/32] incubator-ignite git commit: # ignite-238
Date Fri, 13 Feb 2015 15:57:15 GMT
Repository: incubator-ignite
Updated Branches:
  refs/heads/ignite-230 5b65f983f -> 71091f07e


# ignite-238


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

Branch: refs/heads/ignite-230
Commit: f636bb1593afe78e1af8674097ad7ccb8904f4f8
Parents: 210bd2e
Author: sboikov <sboikov@gridgain.com>
Authored: Thu Feb 12 15:18:42 2015 +0300
Committer: sboikov <sboikov@gridgain.com>
Committed: Thu Feb 12 16:53:40 2015 +0300

----------------------------------------------------------------------
 .../apache/ignite/cache/CacheProjection.java    |   4 +-
 .../processors/cache/GridCacheAdapter.java      |  28 +++--
 .../processors/cache/GridCacheEntryEx.java      |   7 +-
 .../processors/cache/GridCacheMapEntry.java     |  36 ++++--
 .../cache/GridCacheProjectionImpl.java          |   8 +-
 .../processors/cache/GridCacheProxyImpl.java    |   8 +-
 .../cache/IgniteCacheExpiryPolicy.java          |   6 +
 .../processors/cache/IgniteCacheProxy.java      |   2 +-
 .../dht/atomic/GridDhtAtomicCache.java          |   5 +
 .../cache/query/GridCacheQueryManager.java      | 109 +++++++++++++++----
 .../processors/cache/GridCacheTestEntryEx.java  |   8 +-
 .../IgniteCacheExpiryPolicyAbstractTest.java    |  20 ++++
 12 files changed, 194 insertions(+), 47 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f636bb15/modules/core/src/main/java/org/apache/ignite/cache/CacheProjection.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cache/CacheProjection.java b/modules/core/src/main/java/org/apache/ignite/cache/CacheProjection.java
index b6e4dc1..f156b3f 100644
--- a/modules/core/src/main/java/org/apache/ignite/cache/CacheProjection.java
+++ b/modules/core/src/main/java/org/apache/ignite/cache/CacheProjection.java
@@ -476,10 +476,12 @@ public interface CacheProjection<K, V> extends Iterable<Cache.Entry<K,
V>> {
     /**
      * @param key Key.
      * @param peekModes Peek modes.
+     * @param plc Expiry policy if TTL should be updated.
      * @return Value.
      * @throws IgniteCheckedException If failed.
      */
-    @Nullable public V localPeek(K key, CachePeekMode[] peekModes) throws IgniteCheckedException;
+    @Nullable public V localPeek(K key, CachePeekMode[] peekModes, @Nullable IgniteCacheExpiryPolicy
plc)
+        throws IgniteCheckedException;
 
     /**
      * @param peekModes Peek modes.

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f636bb15/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
index 6f75e32..0cd8886 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
@@ -712,7 +712,11 @@ public abstract class GridCacheAdapter<K, V> implements GridCache<K,
V>,
 
     /** {@inheritDoc} */
     @SuppressWarnings("ForLoopReplaceableByForEach")
-    @Nullable @Override public V localPeek(K key, CachePeekMode[] peekModes) throws IgniteCheckedException
{
+    @Nullable @Override public V localPeek(K key,
+        CachePeekMode[] peekModes,
+        @Nullable IgniteCacheExpiryPolicy plc)
+        throws IgniteCheckedException
+    {
         A.notNull(key, "key");
 
         if (keyCheck)
@@ -783,7 +787,7 @@ public abstract class GridCacheAdapter<K, V> implements GridCache<K,
V>,
                         (ctx.isNear() ? ctx.near().dht().peekEx(key) : peekEx(key));
 
                     if (e != null) {
-                        val = e.peek(modes.heap, modes.offheap, modes.swap, topVer);
+                        val = e.peek(modes.heap, modes.offheap, modes.swap, topVer, plc);
 
                         modes.offheap = false;
                         modes.swap = false;
@@ -799,7 +803,7 @@ public abstract class GridCacheAdapter<K, V> implements GridCache<K,
V>,
                 }
             }
             else
-                val = localCachePeek0(key, modes.heap, modes.offheap, modes.swap);
+                val = localCachePeek0(key, modes.heap, modes.offheap, modes.swap, plc);
 
             if (ctx.portableEnabled())
                 val = (V)ctx.unwrapPortableIfNeeded(val, ctx.keepPortable());
@@ -819,11 +823,16 @@ public abstract class GridCacheAdapter<K, V> implements GridCache<K,
V>,
      * @param heap Read heap flag.
      * @param offheap Read offheap flag.
      * @param swap Read swap flag.
+     * @param plc Optional expiry policy.
      * @return Value.
      * @throws GridCacheEntryRemovedException If entry removed.
      * @throws IgniteCheckedException If failed.
      */
-    @Nullable private V localCachePeek0(K key, boolean heap, boolean offheap, boolean swap)
+    @Nullable private V localCachePeek0(K key,
+        boolean heap,
+        boolean offheap,
+        boolean swap,
+        IgniteCacheExpiryPolicy plc)
         throws GridCacheEntryRemovedException, IgniteCheckedException {
         assert ctx.isLocal();
         assert heap || offheap || swap;
@@ -832,7 +841,7 @@ public abstract class GridCacheAdapter<K, V> implements GridCache<K,
V>,
             GridCacheEntryEx<K, V> e = peekEx(key);
 
             if (e != null)
-                return e.peek(heap, offheap, swap, -1);
+                return e.peek(heap, offheap, swap, -1, plc);
         }
 
         if (offheap || swap) {
@@ -2117,9 +2126,9 @@ public abstract class GridCacheAdapter<K, V> implements GridCache<K,
V>,
                                 filter,
                                 expiry);
 
-                            GridCacheVersion ver = entry.version();
-
                             if (val == null) {
+                                GridCacheVersion ver = entry.version();
+
                                 if (misses == null)
                                     misses = new GridLeanMap<>();
 
@@ -6080,6 +6089,11 @@ public abstract class GridCacheAdapter<K, V> implements GridCache<K,
V>,
         }
 
         /** {@inheritDoc} */
+        @Override public synchronized boolean readyToFlush(int cnt) {
+            return (entries != null && entries.size() > cnt) || (rdrsMap != null
&& rdrsMap.size() > cnt);
+        }
+
+        /** {@inheritDoc} */
         @Override public String toString() {
             return S.toString(GetExpiryPolicy.class, this);
         }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f636bb15/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java
index 145627f..ec12281 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java
@@ -607,11 +607,16 @@ public interface GridCacheEntryEx<K, V> {
      * @param offheap Read from offheap flag.
      * @param swap Read from swap flag.
      * @param topVer Topology version.
+     * @param plc Expiry policy if TTL should be updated.
      * @return Value.
      * @throws GridCacheEntryRemovedException If entry has been removed.
      * @throws IgniteCheckedException If failed.
      */
-    @Nullable public V peek(boolean heap, boolean offheap, boolean swap, long topVer)
+    @Nullable public V peek(boolean heap,
+        boolean offheap,
+        boolean swap,
+        long topVer,
+        @Nullable IgniteCacheExpiryPolicy plc)
         throws GridCacheEntryRemovedException, IgniteCheckedException;
 
     /**

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f636bb15/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
index c20cf03..69a95a1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
@@ -2761,14 +2761,15 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K,
V>
     @Nullable @Override public V peek(boolean heap,
         boolean offheap,
         boolean swap,
-        long topVer)
+        long topVer,
+        @Nullable IgniteCacheExpiryPolicy expiryPlc)
         throws GridCacheEntryRemovedException, IgniteCheckedException
     {
         assert heap || offheap || swap;
 
         try {
             if (heap) {
-                GridTuple<V> val = peekGlobal(false, topVer, null);
+                GridTuple<V> val = peekGlobal(false, topVer, null, expiryPlc);
 
                 if (val != null)
                     return val.get();
@@ -2851,13 +2852,13 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K,
V>
                 return peekTx(failFast, filter, tx);
 
             case GLOBAL:
-                return peekGlobal(failFast, topVer, filter);
+                return peekGlobal(failFast, topVer, filter, null);
 
             case NEAR_ONLY:
-                return peekGlobal(failFast, topVer, filter);
+                return peekGlobal(failFast, topVer, filter, null);
 
             case PARTITIONED_ONLY:
-                return peekGlobal(failFast, topVer, filter);
+                return peekGlobal(failFast, topVer, filter, null);
 
             case SMART:
                 /*
@@ -2871,7 +2872,7 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K,
V>
                  * may have enlisted into the same transaction and that's why we pass 'true'
                  * to 'e.peek(true)' method in this case.
                  */
-                return tx == null || tx.state() != ACTIVE ? peekGlobal(failFast, topVer,
filter) :
+                return tx == null || tx.state() != ACTIVE ? peekGlobal(failFast, topVer,
filter, null) :
                     peekTxThenGlobal(failFast, filter, tx);
 
             case SWAP:
@@ -2962,7 +2963,7 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K,
V>
 
         long topVer = tx == null ? cctx.affinity().affinityTopologyVersion() : tx.topologyVersion();
 
-        return peekGlobal(failFast, topVer, filter);
+        return peekGlobal(failFast, topVer, filter, null);
     }
 
     /**
@@ -2982,14 +2983,18 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K,
V>
      * @param failFast Fail fast flag.
      * @param topVer Topology version.
      * @param filter Filter.
+     * @param expiryPlc Optional expiry policy.
      * @return Peeked value.
      * @throws GridCacheFilterFailedException If filter failed.
      * @throws GridCacheEntryRemovedException If entry got removed.
      * @throws IgniteCheckedException If unexpected cache failure occurred.
      */
     @SuppressWarnings({"RedundantTypeArguments"})
-    @Nullable private GridTuple<V> peekGlobal(boolean failFast, long topVer,
-        IgnitePredicate<Cache.Entry<K, V>>[] filter)
+    @Nullable private GridTuple<V> peekGlobal(boolean failFast,
+        long topVer,
+        IgnitePredicate<Cache.Entry<K, V>>[] filter,
+        @Nullable IgniteCacheExpiryPolicy expiryPlc
+        )
         throws GridCacheEntryRemovedException, GridCacheFilterFailedException, IgniteCheckedException
{
         if (!valid(topVer))
             return null;
@@ -3012,6 +3017,19 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K,
V>
 
                     ver = this.ver;
                     val = rawGetOrUnmarshalUnlocked(false);
+
+                    if (val != null && expiryPlc != null) {
+                        long ttl = expiryPlc.forAccess();
+
+                        if (ttl != CU.TTL_NOT_CHANGED) {
+                            updateTtl(ttl);
+
+                            expiryPlc.ttlUpdated(key(),
+                                getOrMarshalKeyBytes(),
+                                version(),
+                                hasReaders() ? ((GridDhtCacheEntry)this).readers() : null);
+                        }
+                    }
                 }
 
                 if (!cctx.isAll(wrap(), filter))

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f636bb15/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProjectionImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProjectionImpl.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProjectionImpl.java
index 9986be7..f074c77 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProjectionImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProjectionImpl.java
@@ -1011,8 +1011,12 @@ public class GridCacheProjectionImpl<K, V> implements GridCacheProjectionEx<K,
V
     }
 
     /** {@inheritDoc} */
-    @Nullable @Override public V localPeek(K key, CachePeekMode[] peekModes) throws IgniteCheckedException
{
-        return cache.localPeek(key, peekModes);
+    @Nullable @Override public V localPeek(K key,
+        CachePeekMode[] peekModes,
+        @Nullable IgniteCacheExpiryPolicy plc)
+        throws IgniteCheckedException
+    {
+        return cache.localPeek(key, peekModes, plc);
     }
 
     /** {@inheritDoc} */

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f636bb15/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java
index 542fd19..653d2a6 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java
@@ -1220,11 +1220,15 @@ public class GridCacheProxyImpl<K, V> implements GridCacheProxy<K,
V>, Externali
     }
 
     /** {@inheritDoc} */
-    @Nullable @Override public V localPeek(K key, CachePeekMode[] peekModes) throws IgniteCheckedException
{
+    @Nullable @Override public V localPeek(K key,
+        CachePeekMode[] peekModes,
+        @Nullable IgniteCacheExpiryPolicy plc)
+        throws IgniteCheckedException
+    {
         GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
         try {
-            return delegate.localPeek(key, peekModes);
+            return delegate.localPeek(key, peekModes, plc);
         }
         finally {
             gate.leave(prev);

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f636bb15/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheExpiryPolicy.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheExpiryPolicy.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheExpiryPolicy.java
index b55d472..deb1d29 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheExpiryPolicy.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheExpiryPolicy.java
@@ -63,6 +63,12 @@ public interface IgniteCacheExpiryPolicy {
     public void reset();
 
     /**
+     * @param cnt Entries count.
+     * @return {@code True} if number of entries or readers is greater than given number.
+     */
+    public boolean readyToFlush(int cnt);
+
+    /**
      * @return Entries with TTL updated on access.
      */
     @Nullable public Map<Object, IgniteBiTuple<byte[], GridCacheVersion>> entries();

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f636bb15/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java
index 9ef4716..db9fbe0 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java
@@ -279,7 +279,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K,
V
         GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
         try {
-            return delegate.localPeek(key, peekModes);
+            return delegate.localPeek(key, peekModes, null);
         }
         catch (IgniteCheckedException e) {
             throw cacheException(e);

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f636bb15/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
index b93fed7..b07c4c7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
@@ -3028,6 +3028,11 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K,
V> {
         }
 
         /** {@inheritDoc} */
+        @Override public boolean readyToFlush(int cnt) {
+            return (entries != null && entries.size() > cnt) || (rdrsMap != null
&& rdrsMap.size() > cnt);
+        }
+
+        /** {@inheritDoc} */
         @Override public String toString() {
             return S.toString(UpdateExpiryPolicy.class, this);
         }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f636bb15/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java
index 985dba9..7fb9d63 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java
@@ -44,6 +44,7 @@ import org.jdk8.backport.*;
 import org.jetbrains.annotations.*;
 
 import javax.cache.*;
+import javax.cache.expiry.*;
 import java.io.*;
 import java.sql.*;
 import java.util.*;
@@ -755,9 +756,17 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
 
         injectResources(keyValFilter);
 
-        GridIterator<IgniteBiTuple<K, V>> heapIt = new GridIteratorAdapter<IgniteBiTuple<K,
V>>() {
+        final CachePeekMode[] peekModes = {CachePeekMode.ONHEAP};
+
+        final GridDhtCacheAdapter dht = cctx.isLocal() ? null : (cctx.isNear() ? cctx.near().dht()
: cctx.dht());
+
+        final ExpiryPolicy plc = cctx.expiry();
+
+        final GridCloseableIteratorAdapter<IgniteBiTuple<K, V>> heapIt = new
GridCloseableIteratorAdapter<IgniteBiTuple<K, V>>() {
             private IgniteBiTuple<K, V> next;
 
+            private IgniteCacheExpiryPolicy expiryPlc = cctx.cache().accessExpiryPolicy(plc);
+
             private Iterator<K> iter = qry.includeBackups() || cctx.isReplicated()
?
                 prj.keySet().iterator() : prj.primaryKeySet().iterator();
 
@@ -765,11 +774,11 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
                 advance();
             }
 
-            @Override public boolean hasNextX() {
+            @Override public boolean onHasNext() {
                 return next != null;
             }
 
-            @Override public IgniteBiTuple<K, V> nextX() {
+            @Override public IgniteBiTuple<K, V> onNext() {
                 if (next == null)
                     throw new NoSuchElementException();
 
@@ -780,10 +789,6 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
                 return next0;
             }
 
-            @Override public void removeX() {
-                throw new UnsupportedOperationException();
-            }
-
             private void advance() {
                 IgniteBiTuple<K, V> next0 = null;
 
@@ -792,7 +797,23 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
 
                     K key = iter.next();
 
-                    V val = prj.peek(key);
+                    V val;
+
+                    try {
+                        val = prj.localPeek(key, peekModes, expiryPlc);
+                    }
+                    catch (IgniteCheckedException e) {
+                        if (log.isDebugEnabled())
+                            log.debug("Failed to peek value: " + e);
+
+                        val = null;
+                    }
+
+                    if (dht != null && expiryPlc != null && expiryPlc.readyToFlush(100))
{
+                        dht.sendTtlUpdateRequest(expiryPlc);
+
+                        expiryPlc = cctx.cache().accessExpiryPolicy(plc);
+                    }
 
                     if (val != null) {
                         next0 = F.t(key, val);
@@ -807,6 +828,21 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
                 next = next0 != null ?
                     new IgniteBiTuple<>(next0.getKey(), next0.getValue()) :
                     null;
+
+                if (next == null)
+                    sendTtlUpdate();
+            }
+
+            @Override protected void onClose() throws IgniteCheckedException {
+                sendTtlUpdate();
+            }
+
+            private void sendTtlUpdate() {
+                if (dht != null && expiryPlc != null) {
+                    dht.sendTtlUpdateRequest(expiryPlc);
+
+                    expiryPlc = null;
+                }
             }
 
             private boolean checkPredicate(Map.Entry<K, V> e) {
@@ -850,6 +886,10 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
             @Override protected void onRemove() {
                 it.remove();
             }
+
+            @Override protected void onClose() throws IgniteCheckedException {
+                heapIt.close();
+            }
         };
     }
 
@@ -998,6 +1038,10 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
 
             boolean rmvRes = true;
 
+            final boolean statsEnabled = cctx.config().isStatisticsEnabled();
+
+            final boolean readEvt = cctx.gridEvents().isRecordable(EVT_CACHE_QUERY_OBJECT_READ);
+
             try {
                 // Preparing query closures.
                 IgnitePredicate<Cache.Entry<Object, Object>> prjFilter = qryInfo.projectionPredicate();
@@ -1054,6 +1098,8 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
                 boolean metaSent = false;
 
                 while (!Thread.currentThread().isInterrupted() && it.hasNext()) {
+                    long start = statsEnabled ? System.nanoTime() : 0L;
+
                     Object row = it.next();
 
                     // Query is cancelled.
@@ -1063,7 +1109,15 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
                         break;
                     }
 
-                    if (cctx.gridEvents().isRecordable(EVT_CACHE_QUERY_OBJECT_READ)) {
+                    if (statsEnabled) {
+                        CacheMetricsImpl metrics = cctx.cache().metrics0();
+
+                        metrics.onRead(true);
+
+                        metrics.addGetTimeNanos(System.nanoTime() - start);
+                    }
+
+                    if (readEvt) {
                         cctx.gridEvents().record(new CacheQueryReadEvent<K, V>(
                             cctx.localNode(),
                             "SQL fields query result set row read.",
@@ -1211,7 +1265,13 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
 
                 long topVer = cctx.affinity().affinityTopologyVersion();
 
+                final boolean statsEnabled = cctx.config().isStatisticsEnabled();
+
+                final boolean readEvt = cctx.gridEvents().isRecordable(EVT_CACHE_QUERY_OBJECT_READ);
+
                 while (!Thread.currentThread().isInterrupted() && iter.hasNext())
{
+                    long start = statsEnabled ? System.nanoTime() : 0L;
+
                     IgniteBiTuple<K, V> row = iter.next();
 
                     // Query is cancelled.
@@ -1248,9 +1308,17 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
                         continue;
                     }
 
-                    switch (type) {
-                        case SQL:
-                            if (cctx.gridEvents().isRecordable(EVT_CACHE_QUERY_OBJECT_READ))
{
+                    if (statsEnabled) {
+                        CacheMetricsImpl metrics = cctx.cache().metrics0();
+
+                        metrics.onRead(true);
+
+                        metrics.addGetTimeNanos(System.nanoTime() - start);
+                    }
+
+                    if (readEvt) {
+                        switch (type) {
+                            case SQL:
                                 cctx.gridEvents().record(new CacheQueryReadEvent<>(
                                     cctx.localNode(),
                                     "SQL query entry read.",
@@ -1268,12 +1336,10 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
                                     val,
                                     null,
                                     null));
-                            }
 
-                            break;
+                                break;
 
-                        case TEXT:
-                            if (cctx.gridEvents().isRecordable(EVT_CACHE_QUERY_OBJECT_READ))
{
+                            case TEXT:
                                 cctx.gridEvents().record(new CacheQueryReadEvent<>(
                                     cctx.localNode(),
                                     "Full text query entry read.",
@@ -1291,12 +1357,10 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
                                     val,
                                     null,
                                     null));
-                            }
 
-                            break;
+                                break;
 
-                        case SCAN:
-                            if (cctx.gridEvents().isRecordable(EVT_CACHE_QUERY_OBJECT_READ))
{
+                            case SCAN:
                                 cctx.gridEvents().record(new CacheQueryReadEvent<>(
                                     cctx.localNode(),
                                     "Scan query entry read.",
@@ -1314,9 +1378,9 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
                                     val,
                                     null,
                                     null));
-                            }
 
-                            break;
+                                break;
+                        }
                     }
 
                     Map.Entry<K, V> entry = F.t(key, val);
@@ -1387,6 +1451,7 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
 
     /**
      * @param qryInfo Info.
+     * @param taskName Task name.
      * @return Iterator.
      * @throws IgniteCheckedException In case of error.
      */

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f636bb15/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestEntryEx.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestEntryEx.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestEntryEx.java
index 7b6e335..ffe3cc9 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestEntryEx.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestEntryEx.java
@@ -836,8 +836,12 @@ public class GridCacheTestEntryEx<K, V> extends GridMetadataAwareAdapter
impleme
     }
 
     /** {@inheritDoc} */
-    @Nullable @Override public V peek(boolean heap, boolean offheap, boolean swap, long topVer)
-        throws GridCacheEntryRemovedException, IgniteCheckedException {
+    @Nullable @Override public V peek(boolean heap,
+        boolean offheap,
+        boolean swap,
+        long topVer,
+        @Nullable IgniteCacheExpiryPolicy plc)
+    {
         return null;
     }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f636bb15/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheExpiryPolicyAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheExpiryPolicyAbstractTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheExpiryPolicyAbstractTest.java
index 59e8bc4..6616958 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheExpiryPolicyAbstractTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheExpiryPolicyAbstractTest.java
@@ -30,6 +30,7 @@ import org.apache.ignite.testframework.*;
 import org.apache.ignite.transactions.*;
 import org.jetbrains.annotations.*;
 
+import javax.cache.*;
 import javax.cache.configuration.*;
 import javax.cache.expiry.*;
 import javax.cache.processor.*;
@@ -307,6 +308,25 @@ public abstract class IgniteCacheExpiryPolicyAbstractTest extends IgniteCacheAbs
                 txGetAll(txMode);
             }
         }
+
+        IgniteCache<Integer, Integer> cache = jcache(0);
+
+        Collection<Integer> putKeys = keys();
+
+        for (final Integer key : putKeys)
+            cache.put(key, key);
+
+        Iterator<Cache.Entry<Integer, Integer>> it = cache.iterator();
+
+        List<Integer> itKeys = new ArrayList<>();
+
+        while (it.hasNext())
+            itKeys.add(it.next().getKey());
+
+        assertTrue(itKeys.size() >= putKeys.size());
+
+        for (Integer key : itKeys)
+            checkTtl(key, 62_000L, true);
     }
 
     /**


Mime
View raw message