#1211 - ENH: Support L2 cache complex natural keys include findList() queries

src/main/java/io/ebean/CacheMode.java

@@ -22,6 +22,11 @@ public enum CacheMode {
    */
   ON(true, true),
 
+  /**
+   * Only used for bean caching. We automatically use the cache for findOne() but not findList().
+   */
+  AUTO(true, true),
+
   /**
    * Do not read from cache, but write retrieved value to cache.
    * Use this, if you want to get the fresh value from database and a CacheMode.ON query will follow.

src/main/java/io/ebean/cache/ServerCache.java

@@ -1,5 +1,9 @@
 package io.ebean.cache;
 
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Set;
+
 /**
  * Represents part of the "L2" server side cache.
  * <p>
@@ -13,6 +17,18 @@ package io.ebean.cache;
  */
 public interface ServerCache {
 
+  default Map<Object,Object> getAll(Set<Object> keys){
+
+    Map<Object,Object> map = new LinkedHashMap<>();
+    for (Object key : keys) {
+      Object value = get(key);
+      if (value != null) {
+        map.put(key, value);
+      }
+    }
+    return map;
+  }
+
   /**
    * Return the value given the key.
    */

src/main/java/io/ebeaninternal/api/BeanCacheResult.java

@@ -0,0 +1,54 @@
+package io.ebeaninternal.api;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * The results of bean cache hit.
+ */
+public class BeanCacheResult<T> {
+
+  private List<Entry<T>> list = new ArrayList<>();
+
+  /**
+   * Add an entry.
+   */
+  public void add(T bean, Object key) {
+    list.add(new Entry<>(bean, key));
+  }
+
+  /**
+   * Return the hits.
+   */
+  public List<Entry<T>> hits() {
+    return list;
+  }
+
+  /**
+   * Bean and cache key pair.
+   */
+  static class Entry<T> {
+
+    private final T bean;
+    private final Object key;
+
+    public Entry(T bean, Object key) {
+      this.bean = bean;
+      this.key = key;
+    }
+
+    /**
+     * Return the natural key or id value.
+     */
+    public Object getKey() {
+      return key;
+    }
+
+    /**
+     * Return the bean.
+     */
+    public T getBean() {
+      return bean;
+    }
+  }
+}

src/main/java/io/ebeaninternal/api/NaturalKeyEntry.java

@@ -0,0 +1,66 @@
+package io.ebeaninternal.api;
+
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Natural key entry with name value pairs for each of the properties making up the key.
+ */
+public class NaturalKeyEntry {
+
+  private Map<String,Object> map = new HashMap<>();
+  private Object key;
+  private Object inValue;
+
+  /**
+   * Used when query query just has a series of EQ expressions (no IN clause).
+   */
+  public NaturalKeyEntry(String[] naturalKey, List<NaturalKeyEq> eqList) {
+    this(naturalKey, eqList, null, null);
+  }
+
+  /**
+   * Create when query uses an IN clause.
+   */
+  public NaturalKeyEntry(String[] naturalKey, List<NaturalKeyEq> eqList, String inProperty, Object inValue) {
+    for (NaturalKeyEq eq : eqList) {
+      map.put(eq.property, eq.value);
+    }
+    if (inProperty != null) {
+      map.put(inProperty, inValue);
+      this.inValue = inValue;
+    }
+    this.key = calculateKey(naturalKey);
+  }
+
+
+  private Object calculateKey(String[] naturalKey) {
+
+    if (naturalKey.length == 1) {
+      return map.get(naturalKey[0]);
+    }
+
+    StringBuilder sb = new StringBuilder();
+    for (String key : naturalKey) {
+      sb.append(map.get(key)).append(";");
+    }
+
+    return sb.toString();
+  }
+
+  /**
+   * Return the natural cache key.
+   */
+  public Object key() {
+    return key;
+  }
+
+  /**
+   * Return the inValue (used to remove from IN clause of original query).
+   */
+  public Object getInValue() {
+    return inValue;
+  }
+}

src/main/java/io/ebeaninternal/api/NaturalKeyEq.java

@@ -0,0 +1,15 @@
+package io.ebeaninternal.api;
+
+/**
+ * A property value pair in a natural key lookup.
+ */
+public class NaturalKeyEq {
+
+  final String property;
+  final Object value;
+
+  public NaturalKeyEq(String property, Object value) {
+    this.property = property;
+    this.value = value;
+  }
+}

src/main/java/io/ebeaninternal/api/NaturalKeyQueryData.java

@@ -0,0 +1,170 @@
+package io.ebeaninternal.api;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Collects the data for processing the natural key cache processing.
+ */
+public class NaturalKeyQueryData<T> {
+
+  private final String[] naturalKey;
+
+  private Collection<?> inValues;
+
+  private String inProperty;
+
+  private List<NaturalKeyEq> eqList;
+
+  private NaturalKeySet set;
+
+  private int hitCount;
+
+  public NaturalKeyQueryData(String[] naturalKey) {
+    this.naturalKey = naturalKey;
+  }
+
+  private boolean matchProperty(String propName) {
+
+    for (String key : naturalKey) {
+      if (key.equals(propName)) {
+        return true;
+      }
+    }
+    return false;
+  }
+
+  /**
+   * Match for IN expression. We only allow one IN clause.
+   */
+  public boolean matchIn(String propName, Collection<?> sourceValues) {
+    if (inProperty != null) {
+      // only 1 IN allowed (to project naturalIds)
+      return false;
+    }
+    if (matchProperty(propName)) {
+      this.inProperty = propName;
+      this.inValues = sourceValues;
+      return true;
+    }
+    return false;
+  }
+
+  /**
+   * Match for an EQ expression.
+   */
+  public boolean matchEq(String propName, Object bindValue) {
+    if (matchProperty(propName)) {
+      if (eqList == null) {
+        eqList = new ArrayList<>();
+      }
+      eqList.add(new NaturalKeyEq(propName, bindValue));
+      return true;
+    }
+    return false;
+  }
+
+  /**
+   * Build and return the set of natural keys we will use.
+   */
+  public NaturalKeySet buildKeys() {
+
+    if (!expressionCount() || !matchProperties()) {
+      return null;
+    }
+
+    this.set = new NaturalKeySet();
+    if (inValues == null) {
+      // only one - a findOne()
+      set.add(new NaturalKeyEntry(naturalKey, eqList));
+
+    } else {
+      // a findList() with an IN clause so we project
+      // for every IN value a natural key combination
+      for (Object inValue : inValues) {
+        set.add(new NaturalKeyEntry(naturalKey, eqList, inProperty, inValue));
+      }
+    }
+
+    return set;
+  }
+
+  /**
+   * Return true if the properties match the natural key properties.
+   */
+  private boolean matchProperties() {
+    if (naturalKey.length == 1) {
+      // simple single property case
+      if (inProperty != null) {
+        return inProperty.equals(naturalKey[0]);
+      } else {
+        return eqList.get(0).property.equals(naturalKey[0]);
+      }
+    }
+
+    // multiple properties case
+    Set<String> exprProps = new HashSet<>();
+    if (inProperty != null) {
+      exprProps.add(inProperty);
+    }
+    if (eqList != null) {
+      for (NaturalKeyEq eq : eqList) {
+        exprProps.add(eq.property);
+      }
+    }
+    if (exprProps.size() != naturalKey.length) {
+      return false;
+    }
+    for (String key : naturalKey) {
+      if (!exprProps.remove(key)) {
+        return false;
+      }
+    }
+
+    return exprProps.isEmpty();
+  }
+
+  /**
+   * Check that all the natural key properties are defined.
+   */
+  private boolean expressionCount() {
+
+    int defined = (inValues == null) ? 0 : 1;
+    defined += (eqList == null) ? 0 : eqList.size();
+    return defined == naturalKey.length;
+  }
+
+  /**
+   * Return the number of entries in the IN clause left remaining (to hit the DB with).
+   */
+  public boolean allHits() {
+    return hitCount > 0
+      && hitCount == set.size()
+      && (inValues == null || inValues.isEmpty());
+  }
+
+  /**
+   * Adjust the IN clause removing the hit entry.
+   */
+  public List<T> removeHits(BeanCacheResult<T> cacheResult) {
+
+    List<BeanCacheResult.Entry<T>> hits = cacheResult.hits();
+    this.hitCount = hits.size();
+
+    List<T> beans = new ArrayList<>(hitCount);
+
+    for (BeanCacheResult.Entry<T> hit : hits) {
+      if (inValues != null) {
+        Object naturalKey = hit.getKey();
+        Object inValue = set.getInValue(naturalKey);
+        inValues.remove(inValue);
+      }
+      beans.add(hit.getBean());
+    }
+
+    return beans;
+  }
+}

src/main/java/io/ebeaninternal/api/NaturalKeySet.java

@@ -0,0 +1,31 @@
+package io.ebeaninternal.api;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Set;
+
+public class NaturalKeySet {
+
+
+  private final Map<Object, NaturalKeyEntry> map = new LinkedHashMap<>();
+
+  public NaturalKeySet() {
+  }
+
+  public int size() {
+    return map.size();
+  }
+
+  public void add(NaturalKeyEntry entry) {
+    map.put(entry.key(), entry);
+  }
+
+  public Set<Object> keys() {
+    return map.keySet();
+  }
+
+  public Object getInValue(Object naturalKey) {
+    NaturalKeyEntry entry = map.get(naturalKey);
+    return entry.getInValue();
+  }
+}

src/main/java/io/ebeaninternal/api/SpiExpression.java

@@ -97,4 +97,9 @@ public interface SpiExpression extends Expression {
    * Return the bind Id value if this is a "equal to" expression for the id property.
    */
   Object getIdEqualTo(String idName);
+
+  /**
+   * Check for match to a natural key query returning false if it doesn't match.
+   */
+  boolean naturalKey(NaturalKeyQueryData data);
 }

src/main/java/io/ebeaninternal/api/SpiExpressionRequest.java

@@ -74,5 +74,5 @@ public interface SpiExpressionRequest {
   /**
    * Append IN expression taking into account platform and type support for Multi-value.
    */
-  void appendInExpression(boolean not, Object[] bindValues);
+  void appendInExpression(boolean not, List<Object> bindValues);
 }

src/main/java/io/ebeaninternal/api/SpiQuery.java

@@ -366,6 +366,17 @@ public interface SpiQuery<T> extends Query<T>, TxnProfileEventCodes {
    */
   ManyWhereJoins getManyWhereJoins();
 
+  /**
+   * Reset AUTO mode to OFF for findList(). Expect explicit cache use with findList().
+   */
+  void resetBeanCacheAutoMode();
+
+  /**
+   * Collect natural key data for this query or null if the query does not match
+   * the requirements of natural key lookup.
+   */
+  NaturalKeyQueryData<T> naturalKey();
+
   /**
    * Return a Natural Key bind parameter if supported by this query.
    */
@@ -579,27 +590,30 @@ public interface SpiQuery<T> extends Query<T>, TxnProfileEventCodes {
   boolean hasMaxRowsOrFirstRow();
 
   /**
-   * Return true if the bean cache should be exclude for query or lazy loading.
+   * Return true if the query should GET against bean cache.
    */
-  boolean isExcludeBeanCache();
+  boolean isBeanCacheGet();
 
   /**
-   * Return true if this query should use the bean cache.
-   * It is not skipped and bean caching is supported.
+   * Return true if the query should PUT against the bean cache.
    */
-  boolean isUseBeanCache();
+  boolean isBeanCachePut();
+
+  /**
+   * Return true if the bean cache is being explicitly loaded via RECACHE mode.
+   */
+  boolean isBeanCacheReload();
+
+  /**
+   * Return the cache mode for using the bean cache (Get and Put).
+   */
+  CacheMode getUseBeanCache();
 
   /**
    * Return the cache mode if this query should use/check the query cache.
    */
   CacheMode getUseQueryCache();
 
-  /**
-   * Return true if the beans from this query should be loaded into the bean
-   * cache.
-   */
-  boolean isLoadBeanCache();
-
   /**
    * Return true if the beans returned by this query should be read only.
    */

src/main/java/io/ebeaninternal/server/autotune/service/BaseQueryTuner.java

@@ -144,7 +144,7 @@ public class BaseQueryTuner {
       default:
         // not using autoTune when explicitly loading the l2 bean cache
         // or when using Versions query
-        return !query.isLoadBeanCache() && SpiQuery.TemporalMode.VERSIONS != query.getTemporalMode();
+        return !query.isBeanCacheReload() && SpiQuery.TemporalMode.VERSIONS != query.getTemporalMode();
     }
   }
 

src/main/java/io/ebeaninternal/server/core/CacheOptions.java

@@ -15,7 +15,7 @@ public class CacheOptions {
   private final boolean enableBeanCache;
   private final boolean enableQueryCache;
   private final boolean readOnly;
-  private final String naturalKey;
+  private final String[] naturalKey;
 
   /**
    * Construct for no caching.
@@ -30,7 +30,7 @@ public class CacheOptions {
   /**
    * Construct with cache annotation.
    */
-  public CacheOptions(Cache cache, String naturalKey) {
+  public CacheOptions(Cache cache, String[] naturalKey) {
     enableBeanCache = cache.enableBeanCache();
     enableQueryCache = cache.enableQueryCache();
     readOnly = cache.readOnly();
@@ -61,7 +61,7 @@ public class CacheOptions {
   /**
    * Return the natural key property name.
    */
-  public String getNaturalKey() {
+  public String[] getNaturalKey() {
     return naturalKey;
   }
 }

src/main/java/io/ebeaninternal/server/core/DefaultServer.java

@@ -1150,7 +1150,7 @@ public final class DefaultServer implements SpiServer, SpiEbeanServer {
       }
     }
 
-    if (!query.isUseBeanCache() || (t != null && t.isSkipCache())) {
+    if (!query.isBeanCacheGet() || (t != null && t.isSkipCache())) {
       return null;
     }
 
@@ -1179,7 +1179,7 @@ public final class DefaultServer implements SpiServer, SpiEbeanServer {
 
     SpiQuery<T> spiQuery = (SpiQuery<T>) query;
     spiQuery.setType(Type.BEAN);
-    if (SpiQuery.Mode.NORMAL == spiQuery.getMode() && !spiQuery.isLoadBeanCache()) {
+    if (SpiQuery.Mode.NORMAL == spiQuery.getMode() && !spiQuery.isBeanCacheReload()) {
       // See if we can skip doing the fetch completely by getting the bean from the
       // persistence context or the bean cache
       T bean = findIdCheckPersistenceContextAndCache(t, spiQuery, spiQuery.getId());
@@ -1217,22 +1217,12 @@ public final class DefaultServer implements SpiServer, SpiEbeanServer {
       return findId(query, transaction);
     }
 
-    SpiTransaction t = (SpiTransaction) transaction;
-    if (t == null) {
-      t = currentServerTransaction();
-    }
-    if (t == null || !t.isSkipCache()) {
-      id = spiQuery.getBeanDescriptor().cacheNaturalKeyIdLookup(spiQuery);
-      if (id != null) {
-        T bean = findIdCheckPersistenceContextAndCache(t, spiQuery, id);
-        if (bean != null) {
-          return bean;
-        }
-      }
+    if (transaction == null) {
+      transaction = currentServerTransaction();
     }
 
     // a query that is expected to return either 0 or 1 beans
-    List<T> list = findList(query, t);
+    List<T> list = findList(query, transaction, true);
     return extractUnique(list);
   }
 
@@ -1515,13 +1505,23 @@ public final class DefaultServer implements SpiServer, SpiEbeanServer {
   }
 
   @Override
-  @SuppressWarnings("unchecked")
   public <T> List<T> findList(Query<T> query, Transaction t) {
+    return findList(query, t, false);
+  }
+
+  @SuppressWarnings("unchecked")
+  private <T> List<T> findList(Query<T> query, Transaction t, boolean findOne) {
 
     SpiOrmQueryRequest<T> request = createQueryRequest(Type.LIST, query, t);
-    Object result = request.getFromQueryCache();
-    if (result != null) {
-      return (List<T>) result;
+    if (!findOne) {
+      request.resetBeanCacheAutoMode();
+      Object result = request.getFromQueryCache();
+      if (result != null) {
+        return (List<T>) result;
+      }
+    }
+    if ((t == null || !t.isSkipCache()) && request.getFromBeanCache()) {
+      return request.getBeanCacheHits();
     }
     if (request.isUseDocStore()) {
       return docStore().findList(request);

src/main/java/io/ebeaninternal/server/core/OrmQueryRequest.java

@@ -1,12 +1,14 @@
 package io.ebeaninternal.server.core;
 
 import io.ebean.CacheMode;
+import io.ebean.OrderBy;
 import io.ebean.PersistenceContextScope;
 import io.ebean.QueryIterator;
 import io.ebean.Version;
 import io.ebean.bean.BeanCollection;
 import io.ebean.bean.EntityBean;
 import io.ebean.bean.PersistenceContext;
+import io.ebean.common.BeanList;
 import io.ebean.common.CopyOnFirstWriteList;
 import io.ebean.event.BeanFindController;
 import io.ebean.event.BeanQueryAdapter;
@@ -28,6 +30,9 @@ import io.ebeaninternal.server.deploy.DeployPropertyParserMap;
 import io.ebeaninternal.server.loadcontext.DLoadContext;
 import io.ebeaninternal.server.query.CQueryPlan;
 import io.ebeaninternal.server.query.CancelableQuery;
+import io.ebeaninternal.api.BeanCacheResult;
+import io.ebeaninternal.api.NaturalKeyQueryData;
+import io.ebeaninternal.api.NaturalKeySet;
 import io.ebeaninternal.server.transaction.DefaultPersistenceContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -73,6 +78,8 @@ public final class OrmQueryRequest<T> extends BeanRequest implements BeanQueryRe
 
   private SpiQuerySecondary secondaryQueries;
 
+  private List<T> cacheBeans;
+
   /**
    * Create the InternalQueryRequest.
    */
@@ -495,8 +502,75 @@ public final class OrmQueryRequest<T> extends BeanRequest implements BeanQueryRe
     beanDescriptor.putQueryPlan(queryPlanKey, queryPlan);
   }
 
-  public boolean isUseBeanCache() {
-    return query.isUseBeanCache();
+  @Override
+  public void resetBeanCacheAutoMode() {
+    query.resetBeanCacheAutoMode();
+  }
+
+  public boolean isBeanCachePut() {
+    return query.isBeanCachePut();
+  }
+
+  /**
+   * Merge in prior L2 bean cache hits with the query result.
+   */
+  public void mergeCacheHits(BeanCollection<T> result) {
+
+    if (cacheBeans != null && !cacheBeans.isEmpty()) {
+      for (T hit : cacheBeans) {
+        result.internalAdd(hit);
+      }
+      // resort in memory here after merging the cache hits with the DB hits
+      if (result instanceof BeanList) {
+        OrderBy<T> orderBy = query.getOrderBy();
+        if (orderBy != null) {
+          beanDescriptor.sort(((BeanList<T>)result).getActualList(), orderBy.toStringFormat());
+        }
+      }
+    }
+  }
+
+  @Override
+  public List<T> getBeanCacheHits() {
+    OrderBy<T> orderBy = query.getOrderBy();
+    if (orderBy != null) {
+      beanDescriptor.sort(cacheBeans, orderBy.toStringFormat());
+    }
+    return cacheBeans;
+  }
+
+  @Override
+  public boolean getFromBeanCache() {
+
+    if (!query.isBeanCacheGet()) {
+      return false;
+    }
+
+    // check if the query can use the bean cache
+    // 1. Find by Ids
+    //    - hit beanCache with Ids
+    //    - keep cache beans, ensure query modified to fetch misses
+    //    - query and Load misses into bean cache
+    //    - merge the 2 results and return
+    //
+
+    if (!beanDescriptor.isNaturalKeyCaching()) {
+      return false;
+    }
+
+    NaturalKeyQueryData<T> data = query.naturalKey();
+    if (data != null) {
+      NaturalKeySet naturalKeySet = data.buildKeys();
+      if (naturalKeySet != null) {
+        // use the natural keys to lookup Ids to then hit the bean cache
+        BeanCacheResult<T> cacheResult = beanDescriptor.naturalKeyLookup(persistenceContext, naturalKeySet.keys());
+        // adjust the query (IN clause) based on the cache hits
+        this.cacheBeans = data.removeHits(cacheResult);
+        return data.allHits();
+      }
+    }
+
+    return false;
   }
 
   /**

src/main/java/io/ebeaninternal/server/core/SpiOrmQueryRequest.java

@@ -118,6 +118,24 @@ public interface SpiOrmQueryRequest<T> extends DocQueryRequest<T> {
    */
   <A> A getFromQueryCache();
 
+  /**
+   * Maybe hit the bean cache returning true if everything was obtained from the
+   * cache (that there were no misses).
+   *
+   * Do this for findList() on many natural keys or many Ids.
+   */
+  boolean getFromBeanCache();
+
+  /**
+   * Return the bean cache hits (when all hits / no misses).
+   */
+  List<T> getBeanCacheHits();
+
+  /**
+   * Reset Bean cache mode AUTO - require explicit setting for bean cache use with findList().
+   */
+  void resetBeanCacheAutoMode();
+
   /**
    * Return the Database platform like clause.
    */

src/main/java/io/ebeaninternal/server/deploy/BeanDescriptor.java

@@ -63,6 +63,7 @@ import io.ebeaninternal.server.persist.DmlUtil;
 import io.ebeaninternal.server.query.CQueryPlan;
 import io.ebeaninternal.server.query.CQueryPlanStats.Snapshot;
 import io.ebean.util.SplitName;
+import io.ebeaninternal.api.BeanCacheResult;
 import io.ebeaninternal.server.querydefn.OrmQueryDetail;
 import io.ebeaninternal.server.rawsql.SpiRawSql;
 import io.ebeaninternal.server.text.json.ReadJson;
@@ -93,6 +94,7 @@ import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 
 /**
@@ -1189,6 +1191,13 @@ public class BeanDescriptor<T> implements MetaBeanInfo, BeanType<T> {
     }
   }
 
+  /**
+   * Return the natural key properties.
+   */
+  public String[] getNaturalKey() {
+    return cacheHelp.getNaturalKey();
+  }
+
   /**
    * Return true if there is bean or query caching for this type.
    */
@@ -1204,6 +1213,13 @@ public class BeanDescriptor<T> implements MetaBeanInfo, BeanType<T> {
     return cacheHelp.isBeanCaching();
   }
 
+  /**
+   * Return true if there is a natural key defined for this bean type.
+   */
+  public boolean isNaturalKeyCaching() {
+    return cacheHelp.isNaturalKeyCaching();
+  }
+
   /**
    * Return true if there is query caching for this type of bean.
    */
@@ -1370,10 +1386,10 @@ public class BeanDescriptor<T> implements MetaBeanInfo, BeanType<T> {
   }
 
   /**
-   * Try to hit the cache using the natural key.
+   * Use natural key lookup to hit the bean cache.
    */
-  public Object cacheNaturalKeyIdLookup(SpiQuery<T> query) {
-    return cacheHelp.naturalKeyIdLookup(query);
+  public BeanCacheResult<T> naturalKeyLookup(PersistenceContext context, Set<Object> keys) {
+    return cacheHelp.naturalKeyLookup(context, keys);
   }
 
   public void cacheNaturalKeyPut(Object id, Object newKey) {

src/main/java/io/ebeaninternal/server/deploy/BeanDescriptorCacheHelp.java

@@ -5,7 +5,7 @@ import io.ebean.bean.EntityBean;
 import io.ebean.bean.EntityBeanIntercept;
 import io.ebean.bean.PersistenceContext;
 import io.ebean.cache.ServerCache;
-import io.ebeaninternal.api.SpiQuery;
+import io.ebeaninternal.api.BeanCacheResult;
 import io.ebeaninternal.api.TransactionEventTable.TableIUD;
 import io.ebeaninternal.server.cache.CacheChangeSet;
 import io.ebeaninternal.server.cache.CachedBeanData;
@@ -16,16 +16,18 @@ import io.ebeaninternal.server.cache.SpiCacheManager;
 import io.ebeaninternal.server.core.CacheOptions;
 import io.ebeaninternal.server.core.PersistRequest;
 import io.ebeaninternal.server.core.PersistRequestBean;
-import io.ebeaninternal.server.querydefn.NaturalKeyBindParam;
 import io.ebeaninternal.server.transaction.DefaultPersistenceContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 /**
  * Helper for BeanDescriptor that manages the bean, query and collection caches.
@@ -57,7 +59,7 @@ final class BeanDescriptorCacheHelp<T> {
   private final String cacheName;
 
   private final BeanPropertyAssocOne<?>[] propertiesOneImported;
-  private final String naturalKeyProperty;
+  private final String[] naturalKey;
 
   private final ServerCache beanCache;
   private final ServerCache naturalKeyCache;
@@ -83,7 +85,7 @@ final class BeanDescriptorCacheHelp<T> {
     this.cacheOptions = cacheOptions;
     this.cacheSharableBeans = cacheSharableBeans;
     this.propertiesOneImported = propertiesOneImported;
-    this.naturalKeyProperty = cacheOptions.getNaturalKey();
+    this.naturalKey = cacheOptions.getNaturalKey();
 
     if (!cacheOptions.isEnableQueryCache()) {
       this.queryCache = null;
@@ -155,6 +157,13 @@ final class BeanDescriptorCacheHelp<T> {
     return beanCache != null;
   }
 
+  /**
+   * Return true if there is natural key caching for this type of bean.
+   */
+  boolean isNaturalKeyCaching() {
+    return naturalKeyCache != null;
+  }
+
   /**
    * Return true if there is bean or query caching on this type.
    */
@@ -162,6 +171,13 @@ final class BeanDescriptorCacheHelp<T> {
     return beanCache != null || queryCache != null;
   }
 
+  /**
+   * Return the natural key properties.
+   */
+  String[] getNaturalKey() {
+    return naturalKey;
+  }
+
   CacheOptions getCacheOptions() {
     return cacheOptions;
   }
@@ -314,36 +330,49 @@ final class BeanDescriptorCacheHelp<T> {
   }
 
   /**
-   * Find the bean using the natural key lookup if available.
+   * Use natural keys to hit the bean cache and return resulting hits.
    */
-  Object naturalKeyIdLookup(SpiQuery<T> query) {
+  BeanCacheResult<T> naturalKeyLookup(PersistenceContext context, Set<Object> keys) {
 
-    if (!isNaturalKeyCaching(query.isUseBeanCache())) {
-      // no natural key caching for this query
-      return null;
+    if (context == null) {
+      context = new DefaultPersistenceContext();
     }
 
-    // check if it is a find by unique id (using the natural key)
-    NaturalKeyBindParam keyBindParam = query.getNaturalKeyBindParam();
-    if (keyBindParam == null || !isNaturalKey(keyBindParam.getName())) {
-      // query is not appropriate
-      return null;
-    }
+    // naturalKey -> Id map
+    Map<Object, Object> naturalKeyMap = naturalKeyCache.getAll(keys);
 
-    // try to lookup the id using the natural key
-    Object id = naturalKeyCache.get(keyBindParam.getValue());
     if (natLog.isTraceEnabled()) {
-      natLog.trace(" LOOKUP {}({}) - id:{}", cacheName, keyBindParam.getValue(), id);
+      natLog.trace(" LOOKUP Many {}({}) - hits:{}", cacheName, keys, naturalKeyMap);
     }
-    return id;
-  }
 
-  private boolean isNaturalKeyCaching(Boolean queryUseCache) {
-    return naturalKeyCache != null && (queryUseCache == null || queryUseCache);
-  }
+    BeanCacheResult<T> result = new BeanCacheResult<>();
+    if (naturalKeyMap.isEmpty()) {
+      return result;
+    }
 
-  private boolean isNaturalKey(String propName) {
-    return propName != null && propName.equals(cacheOptions.getNaturalKey());
+    // create reverse id -> natural key map
+    Map<Object, Object> reverseMap = new HashMap<>();
+    for (Map.Entry<Object, Object> entry : naturalKeyMap.entrySet()) {
+      reverseMap.put(entry.getValue(), entry.getKey());
+    }
+
+    Set<Object> ids = new HashSet<>(naturalKeyMap.values());
+    Map<Object, Object> beanDataMap = beanCache.getAll(ids);
+    if (beanLog.isTraceEnabled()) {
+      beanLog.trace("   GET MANY {}({}) - hits:{}", cacheName, ids, beanDataMap.keySet());
+    }
+    // process the hits into beans etc
+    for (Map.Entry<Object, Object> entry : beanDataMap.entrySet()) {
+
+      Object id = entry.getKey();
+      CachedBeanData cachedBeanData = (CachedBeanData) entry.getValue();
+
+      T bean = convertToBean(id, false, context, cachedBeanData);
+      Object naturalKey = reverseMap.get(id);
+      result.add(bean, naturalKey);
+    }
+
+    return result;
   }
 
   /**
@@ -414,8 +443,8 @@ final class BeanDescriptorCacheHelp<T> {
     }
     getBeanCache().put(id, beanData);
 
-    if (naturalKeyProperty != null) {
-      Object naturalKey = beanData.getData(naturalKeyProperty);
+    if (naturalKey != null) {
+      Object naturalKey = calculateNaturalKey(beanData);
       if (naturalKey != null) {
         if (natLog.isDebugEnabled()) {
           natLog.debug(" PUT {}({}, {})", cacheName, naturalKey, id);
@@ -425,6 +454,21 @@ final class BeanDescriptorCacheHelp<T> {
     }
   }
 
+  private Object calculateNaturalKey(CachedBeanData beanData) {
+    if (naturalKey.length == 1) {
+      return beanData.getData(naturalKey[0]);
+    }
+    StringBuilder sb = new StringBuilder();
+    for (String key : naturalKey) {
+      Object val = beanData.getData(key);
+      if (val == null) {
+        return null;
+      }
+      sb.append(val).append(";");
+    }
+    return sb.toString();
+  }
+
   CachedBeanData beanCacheGetData(Object id) {
     return (CachedBeanData) getBeanCache().get(id);
   }
@@ -450,6 +494,10 @@ final class BeanDescriptorCacheHelp<T> {
       }
       return null;
     }
+    return convertToBean(id, readOnly, context, data);
+  }
+
+  private T convertToBean(Object id, Boolean readOnly, PersistenceContext context, CachedBeanData data) {
     if (cacheSharableBeans && !Boolean.FALSE.equals(readOnly)) {
       Object bean = data.getSharableBean();
       if (bean != null) {
@@ -716,7 +764,7 @@ final class BeanDescriptorCacheHelp<T> {
       }
 
       if (updateNaturalKey) {
-        Object oldKey = existingData.getData(naturalKeyProperty);
+        Object oldKey = calculateNaturalKey(existingData);
         if (oldKey != null) {
           if (natLog.isDebugEnabled()) {
             natLog.debug(".. update {} REMOVE({}) - old key for ({})", cacheName, oldKey, id);

src/main/java/io/ebeaninternal/server/deploy/meta/DeployBeanDescriptor.java

@@ -405,17 +405,17 @@ public class DeployBeanDescriptor<T> {
    */
   public void setCache(Cache cache) {
 
-    String naturalKey = null;
-    if (!cache.naturalKey().isEmpty()) {
-      // find the property and mark as natural key property
-      String propName = cache.naturalKey().trim();
-      DeployBeanProperty beanProperty = getBeanProperty(propName);
+    String[] properties = cache.naturalKey();
+    for (String property : properties) {
+      DeployBeanProperty beanProperty = getBeanProperty(property);
       if (beanProperty != null) {
         beanProperty.setNaturalKey();
-        naturalKey = propName;
       }
     }
-    this.cacheOptions = new CacheOptions(cache, naturalKey);
+    if (properties.length == 0) {
+      properties = null;
+    }
+    this.cacheOptions = new CacheOptions(cache, properties);
   }
 
   /**

src/main/java/io/ebeaninternal/server/expression/AbstractExpression.java

@@ -9,6 +9,7 @@ import io.ebeaninternal.server.deploy.BeanDescriptor;
 import io.ebeaninternal.server.el.ElPropertyDeploy;
 import io.ebeaninternal.server.el.ElPropertyValue;
 import io.ebean.util.SplitName;
+import io.ebeaninternal.api.NaturalKeyQueryData;
 
 /**
  * Base class for simple expressions.
@@ -21,6 +22,12 @@ public abstract class AbstractExpression implements SpiExpression {
     this.propName = propName;
   }
 
+  @Override
+  public boolean naturalKey(NaturalKeyQueryData data) {
+    // by default can't use naturalKey cache
+    return false;
+  }
+
   @Override
   public void simplify() {
     // do nothing

src/main/java/io/ebeaninternal/server/expression/DefaultExampleExpression.java

@@ -4,6 +4,7 @@ import io.ebean.ExampleExpression;
 import io.ebean.LikeType;
 import io.ebean.bean.EntityBean;
 import io.ebean.event.BeanQueryRequest;
+import io.ebean.util.SplitName;
 import io.ebeaninternal.api.ManyWhereJoins;
 import io.ebeaninternal.api.SpiExpression;
 import io.ebeaninternal.api.SpiExpressionRequest;
@@ -11,7 +12,7 @@ import io.ebeaninternal.api.SpiExpressionValidation;
 import io.ebeaninternal.server.deploy.BeanDescriptor;
 import io.ebeaninternal.server.deploy.BeanProperty;
 import io.ebeaninternal.server.deploy.BeanPropertyAssocOne;
-import io.ebean.util.SplitName;
+import io.ebeaninternal.api.NaturalKeyQueryData;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -87,6 +88,12 @@ public class DefaultExampleExpression implements SpiExpression, ExampleExpressio
     }
   }
 
+  @Override
+  public boolean naturalKey(NaturalKeyQueryData data) {
+    // can't use naturalKey cache
+    return false;
+  }
+
   @Override
   public void simplify() {
     // do nothing

src/main/java/io/ebeaninternal/server/expression/DefaultExpressionList.java

@@ -27,6 +27,7 @@ import io.ebeaninternal.api.SpiExpressionRequest;
 import io.ebeaninternal.api.SpiExpressionValidation;
 import io.ebeaninternal.api.SpiJunction;
 import io.ebeaninternal.server.deploy.BeanDescriptor;
+import io.ebeaninternal.api.NaturalKeyQueryData;
 
 import java.io.IOException;
 import java.sql.Timestamp;
@@ -123,6 +124,12 @@ public class DefaultExpressionList<T> implements SpiExpressionList<T> {
     return new JunctionExpression<>(Junction.Type.FILTER, this);
   }
 
+  @Override
+  public boolean naturalKey(NaturalKeyQueryData data) {
+    // can't use naturalKey cache
+    return false;
+  }
+
   @Override
   public void simplify() {
     simplifyEntries();

src/main/java/io/ebeaninternal/server/expression/DefaultExpressionRequest.java

@@ -161,7 +161,7 @@ public class DefaultExpressionRequest implements SpiExpressionRequest {
   }
 
   @Override
-  public void appendInExpression(boolean not, Object[] bindValues) {
+  public void appendInExpression(boolean not, List<Object> bindValues) {
     append(binder.getInExpression(not, bindValues));
   }
 }

src/main/java/io/ebeaninternal/server/expression/ExistsQueryExpression.java

@@ -9,6 +9,7 @@ import io.ebeaninternal.api.SpiExpressionValidation;
 import io.ebeaninternal.api.SpiQuery;
 import io.ebeaninternal.server.deploy.BeanDescriptor;
 import io.ebeaninternal.server.query.CQuery;
+import io.ebeaninternal.api.NaturalKeyQueryData;
 
 import java.io.IOException;
 import java.util.List;
@@ -35,6 +36,12 @@ class ExistsQueryExpression implements SpiExpression, UnsupportedDocStoreExpress
     this.subQuery = null;
   }
 
+  @Override
+  public boolean naturalKey(NaturalKeyQueryData data) {
+    // can't use naturalKey cache
+    return false;
+  }
+
   @Override
   public void simplify() {
     // do nothing

src/main/java/io/ebeaninternal/server/expression/InExpression.java

@@ -6,6 +6,7 @@ import io.ebeaninternal.api.SpiExpression;
 import io.ebeaninternal.api.SpiExpressionRequest;
 import io.ebeaninternal.server.el.ElPropertyValue;
 import io.ebeaninternal.server.persist.MultiValueWrapper;
+import io.ebeaninternal.api.NaturalKeyQueryData;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -20,7 +21,7 @@ class InExpression extends AbstractExpression {
 
   private final Collection<?> sourceValues;
 
-  private Object[] bindValues;
+  private List<Object> bindValues;
 
   private boolean multiValueSupported;
 
@@ -36,25 +37,31 @@ class InExpression extends AbstractExpression {
     this.not = not;
   }
 
-  private Object[] values() {
+  private List<Object> values() {
     List<Object> vals = new ArrayList<>(sourceValues.size());
     for (Object sourceValue : sourceValues) {
       NamedParamHelp.valueAdd(vals, sourceValue);
     }
-    return vals.toArray();
+    return vals;
+  }
+
+  @Override
+  public boolean naturalKey(NaturalKeyQueryData data) {
+    // can't use naturalKey cache for NOT IN
+    return !not && data.matchIn(propName, bindValues);
   }
 
   @Override
   public void prepareExpression(BeanQueryRequest<?> request) {
     bindValues = values();
-    if (bindValues.length > 0) {
-      multiValueSupported = request.isMultiValueSupported((bindValues[0]).getClass());
+    if (bindValues.size() > 0) {
+      multiValueSupported = request.isMultiValueSupported((bindValues.get(0)).getClass());
     }
   }
 
   @Override
   public void writeDocQuery(DocQueryContext context) throws IOException {
-    context.writeIn(propName, values(), not);
+    context.writeIn(propName, values().toArray(), not);
   }
 
   @Override
@@ -66,10 +73,10 @@ class InExpression extends AbstractExpression {
     }
 
     if (prop == null) {
-      if (bindValues.length > 0) {
+      if (bindValues.size() > 0) {
         // if we have no property, we wrap them in a multi value wrapper.
         // later the binder will decide, which bind strategy to use.
-        request.addBindValue(new MultiValueWrapper(Arrays.asList(bindValues)));
+        request.addBindValue(new MultiValueWrapper(bindValues));
       }
     } else {
       List<Object> idList = new ArrayList<>();
@@ -89,7 +96,7 @@ class InExpression extends AbstractExpression {
   @Override
   public void addSql(SpiExpressionRequest request) {
 
-    if (bindValues.length == 0) {
+    if (bindValues.isEmpty()) {
       String expr = not ? "1=1" : "1=0";
       request.append(expr);
       return;
@@ -102,7 +109,7 @@ class InExpression extends AbstractExpression {
 
     if (prop != null) {
       request.append(prop.getAssocIdInExpr(propName));
-      String inClause = prop.getAssocIdInValueExpr(not, bindValues.length);
+      String inClause = prop.getAssocIdInValueExpr(not, bindValues.size());
       request.append(inClause);
 
     } else {
@@ -125,7 +132,7 @@ class InExpression extends AbstractExpression {
     builder.append(" ?");
     if (!multiValueSupported) {
       // query plan specific to the number of parameters in the IN clause
-      builder.append(bindValues.length);
+      builder.append(bindValues.size());
     }
     builder.append("]");
   }
@@ -142,11 +149,11 @@ class InExpression extends AbstractExpression {
   @Override
   public boolean isSameByBind(SpiExpression other) {
     InExpression that = (InExpression) other;
-    if (this.bindValues.length != that.bindValues.length) {
+    if (this.bindValues.size() != that.bindValues.size()) {
       return false;
     }
-    for (int i = 0; i < bindValues.length; i++) {
-      if (!bindValues[i].equals(that.bindValues[i])) {
+    for (int i = 0; i < bindValues.size(); i++) {
+      if (!bindValues.get(i).equals(that.bindValues.get(i))) {
         return false;
       }
     }

src/main/java/io/ebeaninternal/server/expression/JunctionExpression.java

@@ -25,6 +25,7 @@ import io.ebeaninternal.api.SpiExpressionRequest;
 import io.ebeaninternal.api.SpiExpressionValidation;
 import io.ebeaninternal.api.SpiJunction;
 import io.ebeaninternal.server.deploy.BeanDescriptor;
+import io.ebeaninternal.api.NaturalKeyQueryData;
 
 import java.io.IOException;
 import java.sql.Timestamp;
@@ -58,6 +59,12 @@ class JunctionExpression<T> implements SpiJunction<T>, SpiExpression, Expression
     this.exprList = exprList;
   }
 
+  @Override
+  public boolean naturalKey(NaturalKeyQueryData data) {
+    // can't use naturalKey cache
+    return false;
+  }
+
   /**
    * Simplify nested expressions where possible.
    * <p>

src/main/java/io/ebeaninternal/server/expression/LogicExpression.java

@@ -8,6 +8,7 @@ import io.ebeaninternal.api.SpiExpression;
 import io.ebeaninternal.api.SpiExpressionRequest;
 import io.ebeaninternal.api.SpiExpressionValidation;
 import io.ebeaninternal.server.deploy.BeanDescriptor;
+import io.ebeaninternal.api.NaturalKeyQueryData;
 
 import java.io.IOException;
 
@@ -56,6 +57,12 @@ abstract class LogicExpression implements SpiExpression {
     this.expTwo = (SpiExpression) expTwo;
   }
 
+  @Override
+  public boolean naturalKey(NaturalKeyQueryData data) {
+    // can't use naturalKey cache
+    return false;
+  }
+
   @Override
   public void simplify() {
     // do nothing

src/main/java/io/ebeaninternal/server/expression/NestedPathWrapperExpression.java

@@ -6,6 +6,7 @@ import io.ebeaninternal.api.SpiExpression;
 import io.ebeaninternal.api.SpiExpressionRequest;
 import io.ebeaninternal.api.SpiExpressionValidation;
 import io.ebeaninternal.server.deploy.BeanDescriptor;
+import io.ebeaninternal.api.NaturalKeyQueryData;
 
 import java.io.IOException;
 
@@ -23,6 +24,12 @@ class NestedPathWrapperExpression implements SpiExpression {
     this.delegate = delegate;
   }
 
+  @Override
+  public boolean naturalKey(NaturalKeyQueryData data) {
+    // can't use naturalKey cache
+    return false;
+  }
+
   @Override
   public void simplify() {
     // do nothing
@@ -59,7 +66,7 @@ class NestedPathWrapperExpression implements SpiExpression {
   @Override
   public void queryPlanHash(StringBuilder builder) {
     builder.append("NestedPath[");
-    if (nestedPath != null){
+    if (nestedPath != null) {
       builder.append("path:").append(nestedPath).append(" ");
     }
     delegate.queryPlanHash(builder);

src/main/java/io/ebeaninternal/server/expression/NonPrepareExpression.java

@@ -2,12 +2,19 @@ package io.ebeaninternal.server.expression;
 
 import io.ebean.event.BeanQueryRequest;
 import io.ebeaninternal.api.SpiExpression;
+import io.ebeaninternal.api.NaturalKeyQueryData;
 
 /**
  * Base abstract expression that does nothing for prepareExpression().
  */
 abstract class NonPrepareExpression implements SpiExpression {
 
+  @Override
+  public boolean naturalKey(NaturalKeyQueryData data) {
+    // can't use naturalKey cache
+    return false;
+  }
+
   @Override
   public void simplify() {
     // do nothing

src/main/java/io/ebeaninternal/server/expression/NoopExpression.java

@@ -6,6 +6,7 @@ import io.ebeaninternal.api.SpiExpression;
 import io.ebeaninternal.api.SpiExpressionRequest;
 import io.ebeaninternal.api.SpiExpressionValidation;
 import io.ebeaninternal.server.deploy.BeanDescriptor;
+import io.ebeaninternal.api.NaturalKeyQueryData;
 
 import java.io.IOException;
 
@@ -16,6 +17,12 @@ class NoopExpression implements SpiExpression {
 
   protected static final NoopExpression INSTANCE = new NoopExpression();
 
+  @Override
+  public boolean naturalKey(NaturalKeyQueryData data) {
+    // can't use naturalKey cache
+    return false;
+  }
+
   @Override
   public void simplify() {
     // do nothing

src/main/java/io/ebeaninternal/server/expression/NotExpression.java

@@ -7,6 +7,7 @@ import io.ebeaninternal.api.SpiExpression;
 import io.ebeaninternal.api.SpiExpressionRequest;
 import io.ebeaninternal.api.SpiExpressionValidation;
 import io.ebeaninternal.server.deploy.BeanDescriptor;
+import io.ebeaninternal.api.NaturalKeyQueryData;
 
 import java.io.IOException;
 
@@ -21,6 +22,12 @@ final class NotExpression implements SpiExpression {
     this.exp = (SpiExpression) exp;
   }
 
+  @Override
+  public boolean naturalKey(NaturalKeyQueryData data) {
+    // can't use naturalKey cache
+    return false;
+  }
+
   @Override
   public void simplify() {
     // do nothing

src/main/java/io/ebeaninternal/server/expression/SimpleExpression.java

@@ -5,6 +5,7 @@ import io.ebean.plugin.ExpressionPath;
 import io.ebeaninternal.api.SpiExpression;
 import io.ebeaninternal.api.SpiExpressionRequest;
 import io.ebeaninternal.server.el.ElPropertyValue;
+import io.ebeaninternal.api.NaturalKeyQueryData;
 
 import java.io.IOException;
 import java.util.Arrays;
@@ -26,6 +27,15 @@ public class SimpleExpression extends AbstractValueExpression {
     return null;
   }
 
+  @Override
+  public boolean naturalKey(NaturalKeyQueryData data) {
+    // can't use naturalKey cache for NOT IN
+    if (type != Op.EQ) {
+      return false;
+    }
+    return data.matchEq(propName, bindValue);
+  }
+
   @Override
   public void writeDocQuery(DocQueryContext context) throws IOException {
     if (type == Op.BETWEEN) {

src/main/java/io/ebeaninternal/server/loadcontext/DLoadBaseContext.java

@@ -39,9 +39,7 @@ public abstract class DLoadBaseContext {
     this.desc = desc;
     this.queryProps = queryProps;
     this.fullPath = parent.getFullPath(path);
-
-    this.hitCache = !parent.isExcludeBeanCache() && desc.isBeanCaching();
-
+    this.hitCache = parent.isBeanCacheGet() && desc.isBeanCaching();
     this.objectGraphNode = parent.getObjectGraphNode(path);
 
     this.queryFetch = queryProps != null && queryProps.isQueryFetch();

src/main/java/io/ebeaninternal/server/loadcontext/DLoadContext.java

@@ -1,5 +1,6 @@
 package io.ebeaninternal.server.loadcontext;
 
+import io.ebean.CacheMode;
 import io.ebean.bean.BeanCollection;
 import io.ebean.bean.CallStack;
 import io.ebean.bean.EntityBeanIntercept;
@@ -41,7 +42,7 @@ public class DLoadContext implements LoadContext {
   private final boolean asDraft;
   private final Timestamp asOf;
   private final Boolean readOnly;
-  private final boolean excludeBeanCache;
+  private final CacheMode useBeanCache;
   private final int defaultBatchSize;
   private final boolean disableLazyLoading;
   private final boolean disableReadAudit;
@@ -74,7 +75,7 @@ public class DLoadContext implements LoadContext {
     this.persistenceContext = persistenceContext;
     this.origin = initOrigin();
     this.defaultBatchSize = 100;
-    this.excludeBeanCache = false;
+    this.useBeanCache = CacheMode.OFF;
     this.asDraft = false;
     this.asOf = null;
     this.readOnly = false;
@@ -107,7 +108,7 @@ public class DLoadContext implements LoadContext {
     this.readOnly = query.isReadOnly();
     this.disableReadAudit = query.isDisableReadAudit();
     this.disableLazyLoading = query.isDisableLazyLoading();
-    this.excludeBeanCache = query.isExcludeBeanCache();
+    this.useBeanCache = query.getUseBeanCache();
     this.useProfiling = query.getProfilingListener() != null;
 
     ObjectGraphNode parentNode = query.getParentNode();
@@ -158,8 +159,8 @@ public class DLoadContext implements LoadContext {
     registerSecondaryNode(many, props);
   }
 
-  protected boolean isExcludeBeanCache() {
-    return excludeBeanCache;
+  protected boolean isBeanCacheGet() {
+    return useBeanCache.isGet();
   }
 
   /**

src/main/java/io/ebeaninternal/server/persist/Binder.java

@@ -229,9 +229,9 @@ public class Binder {
   /**
    * Return the SQL in clause taking into account Multi-value support.
    */
-  public String getInExpression(boolean not, Object[] bindValues) {
-    ScalarType<?> type = getScalarType(bindValues[0].getClass());
-    return multiValueBind.getInExpression(not, type, bindValues.length);
+  public String getInExpression(boolean not, List<Object> bindValues) {
+    ScalarType<?> type = getScalarType(bindValues.get(0).getClass());
+    return multiValueBind.getInExpression(not, type, bindValues.size());
   }
 
   /**

src/main/java/io/ebeaninternal/server/query/DefaultOrmQueryEngine.java

@@ -151,7 +151,7 @@ public class DefaultOrmQueryEngine implements OrmQueryEngine {
 
     SpiQuery<T> query = request.getQuery();
 
-    if (query.isLoadBeanCache()) {
+    if (query.isBeanCachePut()) {
       // load the individual beans into the bean cache
       BeanDescriptor<T> descriptor = request.getBeanDescriptor();
       Collection<T> c = result.getActualDetails();
@@ -160,6 +160,8 @@ public class DefaultOrmQueryEngine implements OrmQueryEngine {
       }
     }
 
+    request.mergeCacheHits(result);
+
     if (!result.isEmpty() && query.getUseQueryCache().isPut()) {
       // load the query result into the query cache
       result.setReadOnly(true);
@@ -189,7 +191,7 @@ public class DefaultOrmQueryEngine implements OrmQueryEngine {
       result = queryEngine.find(request);
     }
 
-    if (result != null && request.isUseBeanCache()) {
+    if (result != null && request.isBeanCachePut()) {
       request.getBeanDescriptor().cacheBeanPut((EntityBean) result);
     }
 

src/main/java/io/ebeaninternal/server/querydefn/DefaultOrmQuery.java

@@ -29,6 +29,7 @@ import io.ebeaninternal.api.BindParams;
 import io.ebeaninternal.api.CQueryPlanKey;
 import io.ebeaninternal.api.HashQuery;
 import io.ebeaninternal.api.ManyWhereJoins;
+import io.ebeaninternal.api.NaturalKeyQueryData;
 import io.ebeaninternal.api.SpiExpression;
 import io.ebeaninternal.api.SpiExpressionList;
 import io.ebeaninternal.api.SpiExpressionValidation;
@@ -193,9 +194,7 @@ public class DefaultOrmQuery<T> implements SpiQuery<T> {
 
   private boolean usageProfiling = true;
 
-  private boolean loadBeanCache;
-
-  private boolean excludeBeanCache;
+  private CacheMode useBeanCache = CacheMode.AUTO;
 
   private CacheMode useQueryCache = CacheMode.OFF;
 
@@ -375,6 +374,7 @@ public class DefaultOrmQuery<T> implements SpiQuery<T> {
   @Override
   public DefaultOrmQuery<T> asDraft() {
     this.temporalMode = TemporalMode.DRAFT;
+    this.useBeanCache = CacheMode.OFF;
     return this;
   }
 
@@ -641,6 +641,28 @@ public class DefaultOrmQuery<T> implements SpiQuery<T> {
     return !distinct && !singleAttribute;
   }
 
+  @Override
+  public NaturalKeyQueryData<T> naturalKey() {
+
+    if (whereExpressions == null) {
+      return null;
+    }
+    String[] naturalKey = beanDescriptor.getNaturalKey();
+    if (naturalKey == null || naturalKey.length == 0) {
+      return null;
+    }
+
+    NaturalKeyQueryData<T> data = new NaturalKeyQueryData<>(naturalKey);
+    for (SpiExpression expression : whereExpressions.getUnderlyingList()) {
+      // must be eq or in
+      if (!expression.naturalKey(data)) {
+        return null;
+      }
+    }
+
+    return data;
+  }
+
   @Override
   public NaturalKeyBindParam getNaturalKeyBindParam() {
     NaturalKeyBindParam namedBind = null;
@@ -685,15 +707,13 @@ public class DefaultOrmQuery<T> implements SpiQuery<T> {
     copy.m2mIncludeJoin = m2mIncludeJoin;
     copy.profilingListener = profilingListener;
 
-//    copy.query = query;
     copy.rootTableAlias = rootTableAlias;
     copy.distinct = distinct;
     copy.sqlDistinct = sqlDistinct;
     copy.timeout = timeout;
     copy.mapKey = mapKey;
     copy.id = id;
-    copy.loadBeanCache = loadBeanCache;
-    copy.excludeBeanCache = excludeBeanCache;
+    copy.useBeanCache = useBeanCache;
     copy.useQueryCache = useQueryCache;
     copy.readOnly = readOnly;
     if (detail != null) {
@@ -833,7 +853,7 @@ public class DefaultOrmQuery<T> implements SpiQuery<T> {
   @Override
   public DefaultOrmQuery<T> setForUpdate(boolean forUpdate) {
     this.forUpdate = (forUpdate) ? ForUpdate.BASE : null;
-    this.excludeBeanCache = true;
+    this.useBeanCache = CacheMode.OFF;
     return this;
   }
 
@@ -854,7 +874,7 @@ public class DefaultOrmQuery<T> implements SpiQuery<T> {
 
   private DefaultOrmQuery<T> setForUpdateWithMode(ForUpdate mode) {
     this.forUpdate = mode;
-    this.excludeBeanCache = true;
+    this.useBeanCache = CacheMode.OFF;
     return this;
   }
 
@@ -1083,29 +1103,40 @@ public class DefaultOrmQuery<T> implements SpiQuery<T> {
   }
 
   @Override
-  public boolean isExcludeBeanCache() {
-    // not using L2 cache for asDraft() query
-    return excludeBeanCache || isAsDraft();
+  public boolean isBeanCachePut() {
+    return beanDescriptor.isBeanCaching() && useBeanCache.isPut();
   }
 
   @Override
-  public boolean isUseBeanCache() {
-    return !isExcludeBeanCache() && beanDescriptor.isBeanCaching();
+  public boolean isBeanCacheGet() {
+    return beanDescriptor.isBeanCaching() && useBeanCache.isGet();
   }
 
   @Override
-  public CacheMode getUseQueryCache() {
-    // not using L2 cache for asDraft() query
-    if (isAsDraft()) {
-      return CacheMode.OFF;
-    } else {
-      return useQueryCache;
+  public boolean isBeanCacheReload() {
+    return CacheMode.RECACHE == useBeanCache;
+  }
+
+  @Override
+  public void resetBeanCacheAutoMode() {
+    if (useBeanCache == CacheMode.AUTO) {
+      useBeanCache = CacheMode.OFF;
     }
   }
 
+  @Override
+  public CacheMode getUseBeanCache() {
+    return useBeanCache;
+  }
+
+  @Override
+  public CacheMode getUseQueryCache() {
+    return useQueryCache;
+  }
+
   @Override
   public DefaultOrmQuery<T> setUseCache(boolean useCache) {
-    this.excludeBeanCache = !useCache;
+    this.useBeanCache = (useCache) ? CacheMode.ON: CacheMode.OFF;
     return this;
   }
 
@@ -1115,15 +1146,9 @@ public class DefaultOrmQuery<T> implements SpiQuery<T> {
     return this;
   }
 
-  @Override
-  public boolean isLoadBeanCache() {
-    // not using L2 cache for asDraft() query
-    return !isAsDraft() && loadBeanCache;
-  }
-
   @Override
   public DefaultOrmQuery<T> setLoadBeanCache(boolean loadBeanCache) {
-    this.loadBeanCache = loadBeanCache;
+    this.useBeanCache = CacheMode.RECACHE;
     return this;
   }