OPENJPA-515 Add support for targeted query via query hints

git-svn-id: https://svn.apache.org/repos/asf/openjpa/trunk@627888 13f79535-47bb-0310-9956-ffa450edef68

openjpa-slice/src/main/java/org/apache/openjpa/slice/DistributedBrokerImpl.java

@@ -50,6 +50,7 @@ public class DistributedBrokerImpl extends FinalizingBrokerImpl {
 	 * operating set. The slice is assigned only if a StateManager has never
 	 * been assigned before.
 	 */
+	@Override
 	public OpenJPAStateManager persist(Object pc, Object id, boolean explicit,
 			OpCallbacks call) {
 		OpenJPAStateManager sm = getStateManager(pc);

openjpa-slice/src/main/java/org/apache/openjpa/slice/DistributedConfiguration.java

@@ -38,11 +38,11 @@ public interface DistributedConfiguration extends OpenJPAConfiguration {
 	/**
 	 * Gets the active slice identifiers. This list is determined by the
 	 * configuration properties either by explicit listing in 
-	 * <code>slice.Names</code> property or by scanning <code>slice.*.*</code>
+	 * <code>openjpa.slice.Names</code> property or by scanning 
-	 * properties.
+	 * <code>openjpa.slice.*.*</code> properties.
 	 * <br> 
 	 * The ordering of the slice identifiers is determined when they are
-	 * specified explicitly in <code>slice.Names</code> property or 
+	 * specified explicitly in <code>openjpa.slice.Names</code> property or 
 	 * ordered alphabetically when found by scanning the properties.
 	 * <br>
 	 * This list always returns the identifiers that are <em>active</em>, slices
@@ -60,7 +60,7 @@ public interface DistributedConfiguration extends OpenJPAConfiguration {
 	/**
 	 * Gets the slices of given status.
 	 * @param statuses list of status flags. If null, returns all slices 
-	 * irrespective of status;
+	 * irrespective of status.
 	 */
 	List<Slice> getSlices(Slice.Status...statuses);
 	

openjpa-slice/src/main/java/org/apache/openjpa/slice/DistributionPolicy.java

@@ -33,13 +33,14 @@ public interface DistributionPolicy {
 	 * Gets the name of the slice where a given instance will be stored.
 	 *  
 	 * @param pc The newly persistent or to-be-merged object. 
-	 * @param slices list of names of the configured slices. The ordering of 
+	 * @param slices list of names of the active slices. The ordering of 
-	 * the list is either explicit (by the <code>slice.Names</code> property)
+	 * the list is either explicit <code>openjpa.slice.Names</code> property
-	 * or implicit i.e. alphabetic if <code>slice.Names</code> is unspecified.  
+	 * or implicit i.e. alphabetic order of available identifiers if 
+	 * <code>openjpa.slice.Names</code> is unspecified.  
 	 * @param context generic persistence context managing the given instance.
 	 * 
 	 * @return identifier of the slice. This name must match one of the
-	 * configured slice names. 
+	 * given slice names. 
 	 * @see DistributedConfiguration#getActiveSliceNames()
 	 */
 	String distribute(Object pc, List<String> slices, Object context);

openjpa-slice/src/main/java/org/apache/openjpa/slice/ProductDerivation.java

@@ -27,21 +27,34 @@ import org.apache.openjpa.slice.jdbc.DistributedJDBCBrokerFactory;
 /**
  * Derives configuration for Slice.
  * Introduces a specialized BrokerFactory aliased as <code>slice</code>.
- * All Slice specific configuration is prefixed as <code>slice.XXX</code>
+ * All Slice specific configuration is prefixed as 
+ * <code>openjpa.slice.*.*</code>
  * 
  * @author Pinaki Poddar 
  *
  */
 public class ProductDerivation extends AbstractProductDerivation implements
 		OpenJPAProductDerivation {
-
+    /**
+     * Prefix for all Slice-specific configuration properties. 
+     */
+    public static final String PREFIX_SLICE = "openjpa.slice";
+    
+    /**
+     * Hint key <code>openjpa.hint.slice.Target </code> to specify a subset of 
+     * slices for query. The value corresponding to the key is comma-separated
+     * list of slice identifiers.
+     *  
+     */
+    public static final String HINT_TARGET  = "openjpa.hint.slice.Target";
+    
 	@SuppressWarnings("unchecked")
 	public void putBrokerFactoryAliases(Map m) {
 		m.put("slice", DistributedJDBCBrokerFactory.class.getName());
 	}
 
 	public String getConfigurationPrefix() {
-		return "openjpa.slice";
+		return PREFIX_SLICE;
 	}
 
 	public int getType() {

openjpa-slice/src/main/java/org/apache/openjpa/slice/Slice.java

@@ -21,7 +21,8 @@ package org.apache.openjpa.slice;
 import org.apache.openjpa.conf.OpenJPAConfiguration;
 
 /**
- * Represents a database slice of immutable logical name.
+ * Represents a database slice of immutable logical name, a configuration and
+ * status.
  * 
  * @author Pinaki Poddar 
  *

openjpa-slice/src/main/java/org/apache/openjpa/slice/jdbc/DistributedJDBCConfigurationImpl.java

@@ -49,6 +49,7 @@ import org.apache.openjpa.lib.util.Localizer;
 import org.apache.openjpa.slice.DistributedBrokerImpl;
 import org.apache.openjpa.slice.DistributionPolicy;
 import org.apache.openjpa.slice.ExecutorServiceValue;
+import org.apache.openjpa.slice.ProductDerivation;
 import org.apache.openjpa.slice.Slice;
 import org.apache.openjpa.util.UserException;
 
@@ -76,10 +77,10 @@ public class DistributedJDBCConfigurationImpl extends JDBCConfigurationImpl
     protected ExecutorServiceValue executorServicePlugin;
     protected PluginValue distributionPolicyPlugin;
 
-    public static final String PREFIX_SLICE = "openjpa.slice.";
-    public static final String PREFIX_OPENJPA = "openjpa.";
-    public static final String REGEX_DOT = "\\.";
     public static final String DOT = ".";
+    public static final String REGEX_DOT = "\\.";
+    public static final String PREFIX_SLICE = ProductDerivation.PREFIX_SLICE + DOT;
+    public static final String PREFIX_OPENJPA = "openjpa.";
     private static Localizer _loc =
             Localizer.forPackage(DistributedJDBCConfigurationImpl.class);
 
@@ -322,10 +323,11 @@ public class DistributedJDBCConfigurationImpl extends JDBCConfigurationImpl
         if (sliceNames.isEmpty()) {
             throw new UserException(_loc.get("slice-none-configured"));
         } 
+        String unit = getPersistenceUnitName(original);
         for (String key : sliceNames) {
             JDBCConfiguration child = new JDBCConfigurationImpl();
             child.fromProperties(createSliceProperties(original, key));
-            child.setId(key);
+            child.setId(unit+DOT+key);
             Slice slice = new Slice(key, child);
             _slices.add(slice);
             if (log.isTraceEnabled())

openjpa-slice/src/main/java/org/apache/openjpa/slice/jdbc/DistributedStoreManager.java

@@ -21,6 +21,7 @@ package org.apache.openjpa.slice.jdbc;
 import java.sql.Connection;
 import java.sql.SQLException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.BitSet;
 import java.util.Collection;
 import java.util.HashMap;
@@ -61,6 +62,7 @@ import org.apache.openjpa.lib.util.Localizer;
 import org.apache.openjpa.meta.ClassMetaData;
 import org.apache.openjpa.meta.FieldMetaData;
 import org.apache.openjpa.slice.DistributionPolicy;
+import org.apache.openjpa.slice.ProductDerivation;
 import org.apache.openjpa.slice.transaction.DistributedNaiveTransaction;
 import org.apache.openjpa.slice.transaction.DistributedTransactionManager;
 import org.apache.openjpa.slice.transaction.NaiveTransactionManager;
@@ -275,7 +277,8 @@ class DistributedStoreManager extends JDBCStoreManager {
             boolean subclasses, FetchConfiguration fetch) {
         ResultObjectProvider[] tmp = new ResultObjectProvider[_slices.size()];
         int i = 0;
-        for (SliceStoreManager slice : _slices) {
+        List<SliceStoreManager> targets = getTargets(fetch);
+        for (SliceStoreManager slice : targets) {
             tmp[i++] = slice.executeExtent(meta, subclasses, fetch);
         }
         return new MergedResultObjectProvider(tmp);
@@ -371,7 +374,8 @@ class DistributedStoreManager extends JDBCStoreManager {
                 return lookup(slice).initialize(sm, state, fetch, edata);
         }
         // not a part of Query result load. Look into the slices till found
-        for (SliceStoreManager slice : _slices) {
+        List<SliceStoreManager> targets = getTargets(fetch);
+        for (SliceStoreManager slice : targets) {
             if (slice.initialize(sm, state, fetch, edata)) {
                 sm.setImplData(slice.getName(), true);
                 return true;
@@ -501,6 +505,31 @@ class DistributedStoreManager extends JDBCStoreManager {
         DistributedConnection con = new DistributedConnection(list);
         return new RefCountConnection(con);
     }
+    
+    /**
+     * Gets the list of slices mentioned as  
+     * {@link ProductDerivation#HINT_TARGET hint} of the given
+     * {@link FetchConfiguration#getHint(String) fetch configuration}. 
+     * 
+     * @return all active slices if a) the hint is not specified or b) a null 
+     * value or c) a non-String or d) matches no active slice.
+     */
+    List<SliceStoreManager> getTargets(FetchConfiguration fetch) {
+        if (fetch == null)
+            return _slices;
+        Object hint = fetch.getHint(ProductDerivation.HINT_TARGET);
+        if (hint == null || !(hint instanceof String)) 
+            return _slices;
+        List<String> targetNames = Arrays.asList(hint.toString().split("\\,"));
+        List<SliceStoreManager> targets = new ArrayList<SliceStoreManager>();
+        for (SliceStoreManager slice : _slices) {
+           if (targetNames.contains(slice.getName()))
+              targets.add(slice);
+           }
+          if (targets.isEmpty())
+            return _slices;
+        return targets;
+    }
 
     private static class Flusher implements Callable<Collection> {
         final SliceStoreManager store;

openjpa-slice/src/main/java/org/apache/openjpa/slice/jdbc/DistributedStoreQuery.java

@@ -29,6 +29,7 @@ import java.util.concurrent.Future;
 import org.apache.openjpa.jdbc.kernel.JDBCStore;
 import org.apache.openjpa.jdbc.kernel.JDBCStoreQuery;
 import org.apache.openjpa.kernel.ExpressionStoreQuery;
+import org.apache.openjpa.kernel.FetchConfiguration;
 import org.apache.openjpa.kernel.OrderingMergedResultObjectProvider;
 import org.apache.openjpa.kernel.QueryContext;
 import org.apache.openjpa.kernel.StoreQuery;
@@ -37,6 +38,7 @@ import org.apache.openjpa.lib.rop.MergedResultObjectProvider;
 import org.apache.openjpa.lib.rop.RangeResultObjectProvider;
 import org.apache.openjpa.lib.rop.ResultObjectProvider;
 import org.apache.openjpa.meta.ClassMetaData;
+import org.apache.openjpa.slice.ProductDerivation;
 import org.apache.openjpa.util.StoreException;
 
 /**
@@ -63,8 +65,14 @@ class DistributedStoreQuery extends JDBCStoreQuery {
     public Executor newDataStoreExecutor(ClassMetaData meta, boolean subs) {
     	ParallelExecutor ex = new ParallelExecutor(this, meta, subs, _parser, 
     			ctx.getCompilation());
+    	
+    	FetchConfiguration fetch = getContext().getFetchConfiguration();
+    	DistributedStoreManager store = (DistributedStoreManager)getContext()
+    	    .getStoreContext().getStoreManager().getInnermostDelegate();
+        List<SliceStoreManager> targets = store.getTargets(fetch);
         for (StoreQuery q:_queries) {
-        	ex.addExecutor(q.newDataStoreExecutor(meta, subs));
+            if (targets.contains(((JDBCStoreQuery)q).getStore()))
+                ex.addExecutor(q.newDataStoreExecutor(meta, subs));
         }
         return ex;
     }

openjpa-slice/src/test/java/org/apache/openjpa/slice/PersistenceTestCase.java

@@ -215,10 +215,8 @@ public abstract class PersistenceTestCase
             if (!meta.isMapped() || meta.isEmbeddedOnly() 
                 || Modifier.isAbstract(meta.getDescribedType().getModifiers()))
                 continue;
-            List all = em.createQuery("SELECT o FROM " + meta.getTypeAlias() + " o").
+            em.createQuery("DELETE FROM " + meta.getTypeAlias() + " o").
-                getResultList();
+                executeUpdate();
-            for (Object pc:all)
-                em.remove(pc);
         }
         em.getTransaction().commit();
         em.close();

openjpa-slice/src/test/java/org/apache/openjpa/slice/TestQuery.java

@@ -18,11 +18,14 @@
  */
 package org.apache.openjpa.slice;
 
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 
 import javax.persistence.EntityManager;
 import javax.persistence.Query;
 
+import org.apache.openjpa.persistence.OpenJPAEntityManager;
 import org.apache.openjpa.slice.SlicePersistence;
 
 public class TestQuery extends SliceTestCase {
@@ -83,6 +86,21 @@ public class TestQuery extends SliceTestCase {
         assertEquals(limit, result.size());
     }
     
+    public void testHint() {
+        List<String> targets = new ArrayList<String>();
+        targets.add("Even");
+        EntityManager em = emf.createEntityManager();
+        em.getTransaction().begin();
+        Query query = em.createQuery("SELECT p FROM PObject p");
+        query.setHint(ProductDerivation.HINT_TARGET, "Even");
+        List result = query.getResultList();
+        for (Object pc : result) {
+            String slice = SlicePersistence.getSlice(pc);
+            assertTrue(targets.contains(slice));
+        }
+        em.getTransaction().rollback();
+    }
+    
     protected String getPersistenceUnitName() {
         return "ordering";
     }