SONAR-3437 Fix use of persistit with 3-Tuple keys. This allow to restore

optimization of loading of measures.

6 files changed, 135 insertions, 162 deletions

diff --git a/sonar-batch/src/main/java/org/sonar/batch/index/Cache.java b/sonar-batch/src/main/java/org/sonar/batch/index/Cache.java
index 39815dcfcb3..6562674a580 100644
--- a/sonar-batch/src/main/java/org/sonar/batch/index/Cache.java
+++ b/sonar-batch/src/main/java/org/sonar/batch/index/Cache.java
@@ -22,6 +22,7 @@ package org.sonar.batch.index;
 import com.google.common.collect.Sets;
 import com.persistit.Exchange;
 import com.persistit.Key;
+import com.persistit.KeyFilter;
 import com.persistit.exception.PersistitException;
 import org.apache.commons.lang.builder.ToStringBuilder;
 
@@ -29,6 +30,7 @@ import javax.annotation.CheckForNull;
 
 import java.io.Serializable;
 import java.util.Iterator;
+import java.util.NoSuchElementException;
 import java.util.Set;
 
 /**
@@ -46,27 +48,27 @@ public class Cache<V extends Serializable> {
     this.exchange = exchange;
   }
 
-  public Cache put(Object key, V value) {
+  public Cache<V> put(Object key, V value) {
     resetKey(key);
     return doPut(value);
   }
 
-  public Cache put(Object firstKey, Object secondKey, V value) {
+  public Cache<V> put(Object firstKey, Object secondKey, V value) {
     resetKey(firstKey, secondKey);
     return doPut(value);
   }
 
-  public Cache put(Object firstKey, Object secondKey, Object thirdKey, V value) {
+  public Cache<V> put(Object firstKey, Object secondKey, Object thirdKey, V value) {
     resetKey(firstKey, secondKey, thirdKey);
     return doPut(value);
   }
 
-  public Cache put(Object[] key, V value) {
+  public Cache<V> put(Object[] key, V value) {
     resetKey(key);
     return doPut(value);
   }
 
-  private Cache doPut(V value) {
+  private Cache<V> doPut(V value) {
     try {
       exchange.getValue().put(value);
       exchange.store();
@@ -111,6 +113,7 @@ public class Cache<V extends Serializable> {
     return doGet();
   }
 
+  @SuppressWarnings("unchecked")
   @CheckForNull
   private V doGet() {
     try {
@@ -189,27 +192,27 @@ public class Cache<V extends Serializable> {
    *
    * @param group The group name.
    */
-  public Cache clear(Object key) {
+  public Cache<V> clear(Object key) {
     resetKey(key);
     return doClear();
   }
 
-  public Cache clear(Object firstKey, Object secondKey) {
+  public Cache<V> clear(Object firstKey, Object secondKey) {
     resetKey(firstKey, secondKey);
     return doClear();
   }
 
-  public Cache clear(Object firstKey, Object secondKey, Object thirdKey) {
+  public Cache<V> clear(Object firstKey, Object secondKey, Object thirdKey) {
     resetKey(firstKey, secondKey, thirdKey);
     return doClear();
   }
 
-  public Cache clear(Object[] key) {
+  public Cache<V> clear(Object[] key) {
     resetKey(key);
     return doClear();
   }
 
-  private Cache doClear() {
+  private Cache<V> doClear() {
     try {
       Key to = new Key(exchange.getKey());
       to.append(Key.AFTER);
@@ -239,7 +242,7 @@ public class Cache<V extends Serializable> {
    * @param group The group.
    * @return The set of cache keys for this group.
    */
-  @SuppressWarnings("unchecked")
+  @SuppressWarnings("rawtypes")
   public Set keySet(Object key) {
     try {
       Set<Object> keys = Sets.newLinkedHashSet();
@@ -256,6 +259,7 @@ public class Cache<V extends Serializable> {
     }
   }
 
+  @SuppressWarnings("rawtypes")
   public Set keySet(Object firstKey, Object secondKey) {
     try {
       Set<Object> keys = Sets.newLinkedHashSet();
@@ -301,7 +305,8 @@ public class Cache<V extends Serializable> {
       exchange.clear();
       exchange.append(firstKey).append(secondKey).append(Key.BEFORE);
       Exchange iteratorExchange = new Exchange(exchange);
-      return new ValueIterable<V>(iteratorExchange, false);
+      KeyFilter filter = new KeyFilter().append(KeyFilter.simpleTerm(firstKey)).append(KeyFilter.simpleTerm(secondKey));
+      return new ValueIterable<V>(iteratorExchange, filter);
     } catch (Exception e) {
       throw new IllegalStateException("Fail to get values from cache " + name, e);
     }
@@ -310,12 +315,13 @@ public class Cache<V extends Serializable> {
   /**
    * Lazy-loading values for a given key
    */
-  public Iterable<V> values(Object key) {
+  public Iterable<V> values(Object firstKey) {
     try {
       exchange.clear();
-      exchange.append(key).append(Key.BEFORE);
+      exchange.append(firstKey).append(Key.BEFORE);
       Exchange iteratorExchange = new Exchange(exchange);
-      return new ValueIterable<V>(iteratorExchange, false);
+      KeyFilter filter = new KeyFilter().append(KeyFilter.simpleTerm(firstKey));
+      return new ValueIterable<V>(iteratorExchange, filter);
     } catch (Exception e) {
       throw new IllegalStateException("Fail to get values from cache " + name, e);
     }
@@ -328,7 +334,8 @@ public class Cache<V extends Serializable> {
     try {
       exchange.clear().append(Key.BEFORE);
       Exchange iteratorExchange = new Exchange(exchange);
-      return new ValueIterable<V>(iteratorExchange, true);
+      KeyFilter filter = new KeyFilter().append(KeyFilter.ALL);
+      return new ValueIterable<V>(iteratorExchange, filter);
     } catch (Exception e) {
       throw new IllegalStateException("Fail to get values from cache " + name, e);
     }
@@ -336,12 +343,14 @@ public class Cache<V extends Serializable> {
 
   public Iterable<Entry<V>> entries() {
     exchange.clear().to(Key.BEFORE);
-    return new EntryIterable(new Exchange(exchange), true);
+    KeyFilter filter = new KeyFilter().append(KeyFilter.ALL);
+    return new EntryIterable<V>(new Exchange(exchange), filter);
   }
 
-  public Iterable<SubEntry<V>> subEntries(Object key) {
-    exchange.clear().append(key).append(Key.BEFORE);
-    return new SubEntryIterable(new Exchange(exchange), false);
+  public Iterable<Entry<V>> entries(Object firstKey) {
+    exchange.clear().append(firstKey).append(Key.BEFORE);
+    KeyFilter filter = new KeyFilter().append(KeyFilter.simpleTerm(firstKey));
+    return new EntryIterable<V>(new Exchange(exchange), filter);
   }
 
   private void resetKey(Object key) {
@@ -373,8 +382,8 @@ public class Cache<V extends Serializable> {
   private static class ValueIterable<T extends Serializable> implements Iterable<T> {
     private final Iterator<T> iterator;
 
-    private ValueIterable(Exchange exchange, boolean deep) {
-      this.iterator = new ValueIterator<T>(exchange, deep);
+    private ValueIterable(Exchange exchange, KeyFilter keyFilter) {
+      this.iterator = new ValueIterator<T>(exchange, keyFilter);
     }
 
     @Override
@@ -385,34 +394,34 @@ public class Cache<V extends Serializable> {
 
   private static class ValueIterator<T extends Serializable> implements Iterator<T> {
     private final Exchange exchange;
-    private final boolean deep;
+    private final KeyFilter keyFilter;
 
-    private ValueIterator(Exchange exchange, boolean deep) {
+    private ValueIterator(Exchange exchange, KeyFilter keyFilter) {
       this.exchange = exchange;
-      this.deep = deep;
+      this.keyFilter = keyFilter;
     }
 
     @Override
     public boolean hasNext() {
       try {
-        return exchange.hasNext(deep);
+        return exchange.hasNext(keyFilter);
       } catch (PersistitException e) {
         throw new IllegalStateException(e);
       }
     }
 
+    @SuppressWarnings("unchecked")
     @Override
     public T next() {
       try {
-        exchange.next(deep);
+        exchange.next(keyFilter);
       } catch (PersistitException e) {
         throw new IllegalStateException(e);
       }
-      T value = null;
       if (exchange.getValue().isDefined()) {
-        value = (T) exchange.getValue().get();
+        return (T) exchange.getValue().get();
       }
-      return value;
+      throw new NoSuchElementException();
     }
 
     @Override
@@ -421,86 +430,11 @@ public class Cache<V extends Serializable> {
     }
   }
 
-  private static class SubEntryIterable<T extends Serializable> implements Iterable<SubEntry<T>> {
-    private final SubEntryIterator<T> it;
-
-    private SubEntryIterable(Exchange exchange, boolean deep) {
-      it = new SubEntryIterator<T>(exchange, deep);
-    }
-
-    @Override
-    public Iterator<SubEntry<T>> iterator() {
-      return it;
-    }
-  }
-
-  private static class SubEntryIterator<T extends Serializable> implements Iterator<SubEntry<T>> {
-    private final Exchange exchange;
-    private final boolean deep;
-
-    private SubEntryIterator(Exchange exchange, boolean deep) {
-      this.exchange = exchange;
-      this.deep = deep;
-    }
-
-    @Override
-    public boolean hasNext() {
-      try {
-        return exchange.next(deep);
-      } catch (PersistitException e) {
-        throw new IllegalStateException(e);
-      }
-    }
-
-    @Override
-    public SubEntry next() {
-      Serializable value = null;
-      if (exchange.getValue().isDefined()) {
-        value = (Serializable) exchange.getValue().get();
-      }
-      Key key = exchange.getKey();
-      return new SubEntry(key.indexTo(-1).decode(), value);
-    }
-
-    @Override
-    public void remove() {
-      // nothing to do
-    }
-  }
-
-  public static class SubEntry<V extends Serializable> {
-    private final Object key;
-    private final V value;
-
-    SubEntry(Object key, V value) {
-      this.key = key;
-      this.value = value;
-    }
-
-    public Object key() {
-      return key;
-    }
-
-    public String keyAsString() {
-      return (String) key;
-    }
-
-    @CheckForNull
-    public V value() {
-      return value;
-    }
-
-    @Override
-    public String toString() {
-      return ToStringBuilder.reflectionToString(this);
-    }
-  }
-
   private static class EntryIterable<T extends Serializable> implements Iterable<Entry<T>> {
     private final EntryIterator<T> it;
 
-    private EntryIterable(Exchange exchange, boolean deep) {
-      it = new EntryIterator<T>(exchange, deep);
+    private EntryIterable(Exchange exchange, KeyFilter keyFilter) {
+      it = new EntryIterator<T>(exchange, keyFilter);
     }
 
     @Override
@@ -511,34 +445,40 @@ public class Cache<V extends Serializable> {
 
   private static class EntryIterator<T extends Serializable> implements Iterator<Entry<T>> {
     private final Exchange exchange;
-    private final boolean deep;
+    private final KeyFilter keyFilter;
 
-    private EntryIterator(Exchange exchange, boolean deep) {
+    private EntryIterator(Exchange exchange, KeyFilter keyFilter) {
       this.exchange = exchange;
-      this.deep = deep;
+      this.keyFilter = keyFilter;
     }
 
     @Override
     public boolean hasNext() {
       try {
-        return exchange.next(deep);
+        return exchange.hasNext(keyFilter);
       } catch (PersistitException e) {
         throw new IllegalStateException(e);
       }
     }
 
+    @SuppressWarnings("unchecked")
     @Override
-    public Entry next() {
-      Serializable value = null;
-      if (exchange.getValue().isDefined()) {
-        value = (Serializable) exchange.getValue().get();
+    public Entry<T> next() {
+      try {
+        exchange.next(keyFilter);
+      } catch (PersistitException e) {
+        throw new IllegalStateException(e);
       }
-      Key key = exchange.getKey();
-      Object[] array = new Object[key.getDepth()];
-      for (int i = 0; i < key.getDepth(); i++) {
-        array[i] = key.indexTo(i - key.getDepth()).decode();
+      if (exchange.getValue().isDefined()) {
+        T value = (T) exchange.getValue().get();
+        Key key = exchange.getKey();
+        Object[] array = new Object[key.getDepth()];
+        for (int i = 0; i < key.getDepth(); i++) {
+          array[i] = key.indexTo(i - key.getDepth()).decode();
+        }
+        return new Entry<T>(array, value);
       }
-      return new Entry(array, value);
+      throw new NoSuchElementException();
     }
 
     @Override
diff --git a/sonar-batch/src/main/java/org/sonar/batch/index/ComponentDataCache.java b/sonar-batch/src/main/java/org/sonar/batch/index/ComponentDataCache.java
index e131e9277e7..16be9501117 100644
--- a/sonar-batch/src/main/java/org/sonar/batch/index/ComponentDataCache.java
+++ b/sonar-batch/src/main/java/org/sonar/batch/index/ComponentDataCache.java
@@ -46,7 +46,7 @@ public class ComponentDataCache implements BatchComponent {
     return data == null ? null : ((StringData) data).data();
   }
 
-  public <D extends Data> Iterable<Cache.SubEntry<D>> entries(String componentKey) {
-    return cache.subEntries(componentKey);
+  public <D extends Data> Iterable<Cache.Entry<D>> entries(String componentKey) {
+    return cache.entries(componentKey);
   }
 }
diff --git a/sonar-batch/src/main/java/org/sonar/batch/index/ComponentDataPersister.java b/sonar-batch/src/main/java/org/sonar/batch/index/ComponentDataPersister.java
index 39b0327fee5..d00e4cdc84d 100644
--- a/sonar-batch/src/main/java/org/sonar/batch/index/ComponentDataPersister.java
+++ b/sonar-batch/src/main/java/org/sonar/batch/index/ComponentDataPersister.java
@@ -20,7 +20,6 @@
 package org.sonar.batch.index;
 
 import org.sonar.api.database.model.Snapshot;
-import org.sonar.core.persistence.BatchSession;
 import org.sonar.core.persistence.DbSession;
 import org.sonar.core.persistence.MyBatis;
 import org.sonar.core.source.db.SnapshotDataDao;
@@ -35,7 +34,7 @@ public class ComponentDataPersister implements ScanPersister {
   private final MyBatis mybatis;
 
   public ComponentDataPersister(ComponentDataCache data, SnapshotCache snapshots,
-                                SnapshotDataDao dao, MyBatis mybatis) {
+    SnapshotDataDao dao, MyBatis mybatis) {
     this.data = data;
     this.snapshots = snapshots;
     this.dao = dao;
@@ -48,13 +47,13 @@ public class ComponentDataPersister implements ScanPersister {
     for (Map.Entry<String, Snapshot> componentEntry : snapshots.snapshots()) {
       String componentKey = componentEntry.getKey();
       Snapshot snapshot = componentEntry.getValue();
-      for (Cache.SubEntry<Data> dataEntry : data.entries(componentKey)) {
+      for (Cache.Entry<Data> dataEntry : data.entries(componentKey)) {
         Data value = dataEntry.value();
         if (value != null) {
           SnapshotDataDto dto = new SnapshotDataDto();
           dto.setSnapshotId(snapshot.getId());
           dto.setResourceId(snapshot.getResourceId());
-          dto.setDataType(dataEntry.keyAsString());
+          dto.setDataType(dataEntry.key()[1].toString());
           dto.setData(value.writeString());
           dao.insert(session, dto);
         }
diff --git a/sonar-batch/src/main/java/org/sonar/batch/index/DefaultIndex.java b/sonar-batch/src/main/java/org/sonar/batch/index/DefaultIndex.java
index dc0bac9be7e..1625b7c0bce 100644
--- a/sonar-batch/src/main/java/org/sonar/batch/index/DefaultIndex.java
+++ b/sonar-batch/src/main/java/org/sonar/batch/index/DefaultIndex.java
@@ -187,7 +187,13 @@ public class DefaultIndex extends SonarIndex {
   public <M> M getMeasures(Resource resource, MeasuresFilter<M> filter) {
     // Reload resource so that effective key is populated
     Resource indexedResource = getResource(resource);
-    Iterable<Measure> unfiltered = measureCache.byResource(indexedResource);
+    Iterable<Measure> unfiltered;
+    if (filter instanceof MeasuresFilters.MetricFilter) {
+      // optimization
+      unfiltered = measureCache.byMetric(indexedResource, ((MeasuresFilters.MetricFilter<M>) filter).filterOnMetricKey());
+    } else {
+      unfiltered = measureCache.byResource(indexedResource);
+    }
     Collection<Measure> all = new ArrayList<Measure>();
     if (unfiltered != null) {
       for (Measure measure : unfiltered) {
diff --git a/sonar-batch/src/main/java/org/sonar/batch/scan/measure/MeasureCache.java b/sonar-batch/src/main/java/org/sonar/batch/scan/measure/MeasureCache.java
index b29b6c83ee0..aaa6adee60d 100644
--- a/sonar-batch/src/main/java/org/sonar/batch/scan/measure/MeasureCache.java
+++ b/sonar-batch/src/main/java/org/sonar/batch/scan/measure/MeasureCache.java
@@ -47,17 +47,21 @@ public class MeasureCache implements BatchComponent {
     return cache.values(r.getEffectiveKey());
   }
 
+  public Iterable<Measure> byMetric(Resource r, String metricKey) {
+    return cache.values(r.getEffectiveKey(), metricKey);
+  }
+
   public MeasureCache put(Resource resource, Measure measure) {
     Preconditions.checkNotNull(resource.getEffectiveKey());
     Preconditions.checkNotNull(measure.getMetricKey());
-    cache.put(resource.getEffectiveKey(), computeMeasureKey(measure), measure);
+    cache.put(resource.getEffectiveKey(), measure.getMetricKey(), computeMeasureKey(measure), measure);
     return this;
   }
 
   public boolean contains(Resource resource, Measure measure) {
     Preconditions.checkNotNull(resource.getEffectiveKey());
     Preconditions.checkNotNull(measure.getMetricKey());
-    return cache.containsKey(resource.getEffectiveKey(), computeMeasureKey(measure));
+    return cache.containsKey(resource.getEffectiveKey(), measure.getMetricKey(), computeMeasureKey(measure));
   }
 
   private static String computeMeasureKey(Measure m) {
diff --git a/sonar-batch/src/test/java/org/sonar/batch/index/CacheTest.java b/sonar-batch/src/test/java/org/sonar/batch/index/CacheTest.java
index cd82fb3693e..f631ef864ec 100644
--- a/sonar-batch/src/test/java/org/sonar/batch/index/CacheTest.java
+++ b/sonar-batch/src/test/java/org/sonar/batch/index/CacheTest.java
@@ -99,44 +99,47 @@ public class CacheTest {
 
     cache.put("europe", "france", "paris");
     cache.put("europe", "italy", "rome");
+    cache.put("asia", "china", "pekin");
     assertThat(cache.get("europe")).isNull();
     assertThat(cache.get("europe", "france")).isEqualTo("paris");
     assertThat(cache.get("europe", "italy")).isEqualTo("rome");
     assertThat(cache.get("europe")).isNull();
     assertThat(cache.keySet("europe")).containsOnly("france", "italy");
-    assertThat(cache.keySet()).containsOnly("europe");
+    assertThat(cache.keySet()).containsOnly("europe", "asia");
     assertThat(cache.containsKey("europe")).isFalse();
     assertThat(cache.containsKey("europe", "france")).isTrue();
     assertThat(cache.containsKey("europe", "spain")).isFalse();
-    assertThat(cache.values()).containsOnly("paris", "rome");
+    assertThat(cache.values()).containsOnly("paris", "rome", "pekin");
     assertThat(cache.values("america")).isEmpty();
     assertThat(cache.values("europe")).containsOnly("paris", "rome");
     assertThat(cache.values("oceania")).isEmpty();
 
     Cache.Entry[] allEntries = Iterables.toArray(cache.entries(), Cache.Entry.class);
-    assertThat(allEntries).hasSize(2);
-    assertThat(allEntries[0].key()).isEqualTo(new String[]{"europe", "france"});
-    assertThat(allEntries[0].value()).isEqualTo("paris");
-    assertThat(allEntries[1].key()).isEqualTo(new String[]{"europe", "italy"});
-    assertThat(allEntries[1].value()).isEqualTo("rome");
-
-    Cache.SubEntry[] subEntries = Iterables.toArray(cache.subEntries("europe"), Cache.SubEntry.class);
+    assertThat(allEntries).hasSize(3);
+    assertThat(allEntries[0].key()).isEqualTo(new String[] {"asia", "china"});
+    assertThat(allEntries[0].value()).isEqualTo("pekin");
+    assertThat(allEntries[1].key()).isEqualTo(new String[] {"europe", "france"});
+    assertThat(allEntries[1].value()).isEqualTo("paris");
+    assertThat(allEntries[2].key()).isEqualTo(new String[] {"europe", "italy"});
+    assertThat(allEntries[2].value()).isEqualTo("rome");
+
+    Cache.Entry[] subEntries = Iterables.toArray(cache.entries("europe"), Cache.Entry.class);
     assertThat(subEntries).hasSize(2);
-    assertThat(subEntries[0].keyAsString()).isEqualTo("france");
+    assertThat(subEntries[0].key()).isEqualTo(new String[] {"europe", "france"});
     assertThat(subEntries[0].value()).isEqualTo("paris");
-    assertThat(subEntries[1].keyAsString()).isEqualTo("italy");
+    assertThat(subEntries[1].key()).isEqualTo(new String[] {"europe", "italy"});
     assertThat(subEntries[1].value()).isEqualTo("rome");
 
     cache.remove("europe", "france");
-    assertThat(cache.values()).containsOnly("rome");
+    assertThat(cache.values()).containsOnly("rome", "pekin");
     assertThat(cache.get("europe", "france")).isNull();
     assertThat(cache.get("europe", "italy")).isEqualTo("rome");
     assertThat(cache.containsKey("europe", "france")).isFalse();
     assertThat(cache.keySet("europe")).containsOnly("italy");
 
     cache.clear("america");
-    assertThat(cache.keySet()).containsOnly("europe");
-    cache.clear("europe");
+    assertThat(cache.keySet()).containsOnly("europe", "asia");
+    cache.clear();
     assertThat(cache.keySet()).isEmpty();
   }
 
@@ -147,47 +150,68 @@ public class CacheTest {
 
     cache.put("europe", "france", "paris", "eiffel tower");
     cache.put("europe", "france", "annecy", "lake");
+    cache.put("europe", "france", "poitiers", "notre dame");
     cache.put("europe", "italy", "rome", "colosseum");
+    cache.put("europe2", "ukrania", "kiev", "dunno");
+    cache.put("asia", "china", "pekin", "great wall");
+    cache.put("america", "us", "new york", "empire state building");
     assertThat(cache.get("europe")).isNull();
     assertThat(cache.get("europe", "france")).isNull();
     assertThat(cache.get("europe", "france", "paris")).isEqualTo("eiffel tower");
     assertThat(cache.get("europe", "france", "annecy")).isEqualTo("lake");
     assertThat(cache.get("europe", "italy", "rome")).isEqualTo("colosseum");
-    assertThat(cache.keySet()).containsOnly("europe");
+    assertThat(cache.keySet()).containsOnly("europe", "asia", "america", "europe2");
     assertThat(cache.keySet("europe")).containsOnly("france", "italy");
-    assertThat(cache.keySet("europe", "france")).containsOnly("annecy", "paris");
+    assertThat(cache.keySet("europe", "france")).containsOnly("annecy", "paris", "poitiers");
     assertThat(cache.containsKey("europe")).isFalse();
     assertThat(cache.containsKey("europe", "france")).isFalse();
     assertThat(cache.containsKey("europe", "france", "annecy")).isTrue();
     assertThat(cache.containsKey("europe", "france", "biarritz")).isFalse();
-    assertThat(cache.values()).containsOnly("eiffel tower", "lake", "colosseum");
+    assertThat(cache.values()).containsOnly("eiffel tower", "lake", "colosseum", "notre dame", "great wall", "empire state building", "dunno");
+    assertThat(cache.values("europe")).containsOnly("eiffel tower", "lake", "colosseum", "notre dame");
+    assertThat(cache.values("europe", "france")).containsOnly("eiffel tower", "lake", "notre dame");
 
     Cache.Entry[] allEntries = Iterables.toArray(cache.entries(), Cache.Entry.class);
-    assertThat(allEntries).hasSize(3);
-    assertThat(allEntries[0].key()).isEqualTo(new String[]{"europe", "france", "annecy"});
-    assertThat(allEntries[0].value()).isEqualTo("lake");
-    assertThat(allEntries[1].key()).isEqualTo(new String[]{"europe", "france", "paris"});
-    assertThat(allEntries[1].value()).isEqualTo("eiffel tower");
-    assertThat(allEntries[2].key()).isEqualTo(new String[]{"europe", "italy", "rome"});
-    assertThat(allEntries[2].value()).isEqualTo("colosseum");
-
-    Cache.SubEntry[] subEntries = Iterables.toArray(cache.subEntries("europe"), Cache.SubEntry.class);
-    assertThat(subEntries).hasSize(2);
-    assertThat(subEntries[0].keyAsString()).isEqualTo("france");
-    assertThat(subEntries[0].value()).isNull();
-    assertThat(subEntries[1].keyAsString()).isEqualTo("italy");
-    assertThat(subEntries[1].value()).isNull();
+    assertThat(allEntries).hasSize(7);
+    assertThat(allEntries[0].key()).isEqualTo(new String[] {"america", "us", "new york"});
+    assertThat(allEntries[0].value()).isEqualTo("empire state building");
+    assertThat(allEntries[1].key()).isEqualTo(new String[] {"asia", "china", "pekin"});
+    assertThat(allEntries[1].value()).isEqualTo("great wall");
+    assertThat(allEntries[2].key()).isEqualTo(new String[] {"europe", "france", "annecy"});
+    assertThat(allEntries[2].value()).isEqualTo("lake");
+    assertThat(allEntries[3].key()).isEqualTo(new String[] {"europe", "france", "paris"});
+    assertThat(allEntries[3].value()).isEqualTo("eiffel tower");
+    assertThat(allEntries[4].key()).isEqualTo(new String[] {"europe", "france", "poitiers"});
+    assertThat(allEntries[4].value()).isEqualTo("notre dame");
+    assertThat(allEntries[5].key()).isEqualTo(new String[] {"europe", "italy", "rome"});
+    assertThat(allEntries[5].value()).isEqualTo("colosseum");
+
+    Cache.Entry[] subEntries = Iterables.toArray(cache.entries("europe"), Cache.Entry.class);
+    assertThat(subEntries).hasSize(4);
+    assertThat(subEntries[0].key()).isEqualTo(new String[] {"europe", "france", "annecy"});
+    assertThat(subEntries[0].value()).isEqualTo("lake");
+    assertThat(subEntries[1].key()).isEqualTo(new String[] {"europe", "france", "paris"});
+    assertThat(subEntries[1].value()).isEqualTo("eiffel tower");
+    assertThat(subEntries[2].key()).isEqualTo(new String[] {"europe", "france", "poitiers"});
+    assertThat(subEntries[2].value()).isEqualTo("notre dame");
+    assertThat(subEntries[3].key()).isEqualTo(new String[] {"europe", "italy", "rome"});
+    assertThat(subEntries[3].value()).isEqualTo("colosseum");
 
     cache.remove("europe", "france", "annecy");
-    assertThat(cache.values()).containsOnly("eiffel tower", "colosseum");
+    assertThat(cache.values()).containsOnly("eiffel tower", "colosseum", "notre dame", "great wall", "empire state building", "dunno");
+    assertThat(cache.values("europe")).containsOnly("eiffel tower", "colosseum", "notre dame");
+    assertThat(cache.values("europe", "france")).containsOnly("eiffel tower", "notre dame");
     assertThat(cache.get("europe", "france", "annecy")).isNull();
     assertThat(cache.get("europe", "italy", "rome")).isEqualTo("colosseum");
     assertThat(cache.containsKey("europe", "france")).isFalse();
 
     cache.clear("europe", "italy");
-    assertThat(cache.values()).containsOnly("eiffel tower");
+    assertThat(cache.values()).containsOnly("eiffel tower", "notre dame", "great wall", "empire state building", "dunno");
 
     cache.clear("europe");
+    assertThat(cache.values()).containsOnly("great wall", "empire state building", "dunno");
+
+    cache.clear();
     assertThat(cache.values()).isEmpty();
   }