Permalink
Browse files

-JDOQL basic range queries are now supported by the plugin.

-plugin.xml contains the extension to signal JDOQL support
  • Loading branch information...
1 parent f1dedb7 commit fa1c55a923a34bd4e01bbb978707282d5136ac81 pedro committed Jun 8, 2010
View
@@ -27,5 +27,15 @@
transactional="false">
</connectionfactory>
</extension>
+ <extension
+ point="org.datanucleus.store_query_query">
+ <query class-name="org.datanucleus.store.cassandra.query.JDOQLQuery" datastore="cassandra" name="javax.jdo.query.JDOQL"/>
+ <query class-name="org.datanucleus.store.cassandra.query.JDOQLQuery" datastore="cassandra" name="JDOQL"/>
+ <query
+ class-name="org.datanucleus.store.cassandra.query.JDOQLQuery"
+ datastore="cassandra"
+ name="cassandra">
+ </query>
+ </extension>
</plugin>
@@ -28,9 +28,11 @@
public class CassandraQuery {
+ public static int search_slice_ratio = 1000; //should come from the properties
+
static List getObjectsOfCandidateType(final ExecutionContext ec,
final CassandraManagedConnection mconn, Class candidateClass,
- boolean subclasses, boolean ignoreCache) {
+ boolean subclasses, boolean ignoreCache,long fromInclNo , long toExclNo) {
List results = new ArrayList();
try {
@@ -55,8 +57,40 @@ static List getObjectsOfCandidateType(final ExecutionContext ec,
}
slice_predicate.setColumn_names(column_names);
- List<KeySlice> result = c.get_range_slice(keyspace, parent,
- slice_predicate, "", "", 3000, ConsistencyLevel.QUORUM);
+ String last_key = "";
+ int number_keys =0;
+ boolean terminated = false;
+
+
+
+ long limit = toExclNo;//(toExclNo<0) ? -1 : (toExclNo-1);
+ List<KeySlice> result = new ArrayList<KeySlice>();
+
+ while (!terminated) {
+ List<KeySlice> keys = c.get_range_slice(keyspace, parent, slice_predicate, last_key, "", search_slice_ratio, ConsistencyLevel.QUORUM);
+
+ if (!keys.isEmpty()) {
+ last_key = keys.get(keys.size() - 1).key;
+ }
+
+ for (KeySlice key : keys) {
+ if(!key.getColumns().isEmpty()){
+ number_keys++;
+ if(number_keys>fromInclNo){
+ result.add(key);
+ }
+
+ }
+ if (number_keys >= limit) {
+ terminated = true;
+ break;
+ }
+
+ }
+ if (keys.size() < search_slice_ratio) {
+ terminated = true;
+ }
+ }
Iterator<KeySlice> iterator = result.iterator();
@@ -74,29 +108,14 @@ public FetchPlan getFetchPlanForLoading() {
}
@Override
- public void fetchNonLoadedFields(
- ObjectProvider sm) {
- sm.replaceNonLoadedFields(acmd
- .getAllMemberPositions(),
- new CassandraFetchFieldManager(
- acmd, sm, keySlice
- .getColumns()));
-
+ public void fetchNonLoadedFields(ObjectProvider sm) {
+ sm.replaceNonLoadedFields(acmd.getAllMemberPositions(),new CassandraFetchFieldManager(acmd, sm, keySlice.getColumns()));
}
@Override
public void fetchFields(ObjectProvider sm) {
- sm.replaceFields(acmd
- .getPKMemberPositions(),
- new CassandraFetchFieldManager(
- acmd, sm, keySlice
- .getColumns()));
- sm.replaceFields(acmd
- .getBasicMemberPositions(clr, ec
- .getMetaDataManager()),
- new CassandraFetchFieldManager(
- acmd, sm, keySlice
- .getColumns()));
+ sm.replaceFields(acmd.getPKMemberPositions(),new CassandraFetchFieldManager(acmd, sm, keySlice.getColumns()));
+ sm.replaceFields(acmd.getBasicMemberPositions(clr, ec.getMetaDataManager()),new CassandraFetchFieldManager(acmd, sm, keySlice.getColumns()));
}
}, ignoreCache, true));
@@ -7,96 +7,120 @@
import java.util.Map;
import org.datanucleus.ObjectManager;
+import org.datanucleus.exceptions.NucleusUserException;
import org.datanucleus.query.evaluator.JDOQLEvaluator;
import org.datanucleus.query.evaluator.JavaQueryEvaluator;
import org.datanucleus.store.ExecutionContext;
-
import org.datanucleus.store.cassandra.CassandraManagedConnection;
import org.datanucleus.store.query.AbstractJDOQLQuery;
import org.datanucleus.util.NucleusLogger;
public class JDOQLQuery extends AbstractJDOQLQuery {
/**
- * Constructs a new query instance that uses the given persistence manager.
- * @param om the associated ExecutiongContext for this query.
- */
- public JDOQLQuery(ExecutionContext ec)
- {
- this(ec, (JDOQLQuery) null);
- }
-
- /**
- * Constructs a new query instance having the same criteria as the given query.
- * @param om The Executing Manager
- * @param q The query from which to copy criteria.
- */
- public JDOQLQuery(ExecutionContext ec, JDOQLQuery q)
- {
- super(ec, q);
- }
-
- /**
- * Constructor for a JDOQL query where the query is specified using the "Single-String" format.
- * @param ec The execution context
- * @param query The query string
- */
- public JDOQLQuery(ExecutionContext ec, String query)
- {
- super(ec, query);
- }
-
-
+ * Constructs a new query instance that uses the given persistence manager.
+ *
+ * @param om
+ * the associated ExecutiongContext for this query.
+ */
+ public JDOQLQuery(ExecutionContext ec) {
+ this(ec, (JDOQLQuery) null);
+ }
+
+ /**
+ * Constructs a new query instance having the same criteria as the given
+ * query.
+ *
+ * @param om
+ * The Executing Manager
+ * @param q
+ * The query from which to copy criteria.
+ */
+ public JDOQLQuery(ExecutionContext ec, JDOQLQuery q) {
+ super(ec, q);
+ }
+
+ /**
+ * Constructor for a JDOQL query where the query is specified using the
+ * "Single-String" format.
+ *
+ * @param ec
+ * The execution context
+ * @param query
+ * The query string
+ */
+ public JDOQLQuery(ExecutionContext ec, String query) {
+ super(ec, query);
+ }
+
@Override
protected Object performExecute(Map parameters) {
-
-
- CassandraManagedConnection mconn = (CassandraManagedConnection) ec.getStoreManager().getConnection(ec);
- try
- {
- long startTime = System.currentTimeMillis();
- if (NucleusLogger.QUERY.isDebugEnabled())
- {
- NucleusLogger.QUERY.debug(LOCALISER.msg("021046", "JDOQL", getSingleStringQuery(), null));
- }
- List candidates = null;
- if (candidateCollection != null)
- {
- candidates = new ArrayList(candidateCollection);
- }
- else if (candidateExtent != null)
- {
- candidates = new ArrayList();
- Iterator iter = candidateExtent.iterator();
- while (iter.hasNext())
- {
- candidates.add(iter.next());
- }
- }
- else
- {
- candidates = CassandraQuery.getObjectsOfCandidateType(ec, mconn, candidateClass, subclasses,
- ignoreCache);
- }
-
- // Apply any result restrictions to the results
- JavaQueryEvaluator resultMapper = new JDOQLEvaluator(this, candidates, compilation,
- parameters, ec.getClassLoaderResolver());
- Collection results = resultMapper.execute(true, true, true, true, true);
-
- if (NucleusLogger.QUERY.isDebugEnabled())
- {
- NucleusLogger.QUERY.debug(LOCALISER.msg("021074", "JDOQL",
- "" + (System.currentTimeMillis() - startTime)));
- }
-
- return results;
- }
- finally
- {
- mconn.release();
- }
- }
+
+ if (range != null && !range.equals("")) {
+ // Range is of the format "from, to"
+ String[] fromTo = range.split(",");
+ if (fromTo.length != 2) {
+ throw new NucleusUserException("Malformed RANGE clause: "
+ + range);
+ }
+
+ try {
+ fromInclNo = Long.parseLong(fromTo[0].trim());
+ } catch (Exception e) {
+ throw new NucleusUserException("Malformed RANGE clause: "
+ + range);
+ }
+
+ try {
+ toExclNo = Long.parseLong(fromTo[1].trim());
+ } catch (Exception e) {
+ throw new NucleusUserException("Malformed RANGE clause: "
+ + range);
+ }
+
+ }
+ //
+
+ CassandraManagedConnection mconn = (CassandraManagedConnection) ec
+ .getStoreManager().getConnection(ec);
+ try {
+ long startTime = System.currentTimeMillis();
+ if (NucleusLogger.QUERY.isDebugEnabled()) {
+ NucleusLogger.QUERY.debug(LOCALISER.msg("021046", "JDOQL",
+ getSingleStringQuery(), null));
+ }
+ List candidates = null;
+ if (candidateCollection != null) {
+ candidates = new ArrayList(candidateCollection);
+ } else if (candidateExtent != null) {
+ candidates = new ArrayList();
+ Iterator iter = candidateExtent.iterator();
+ while (iter.hasNext()) {
+ candidates.add(iter.next());
+ }
+ } else {
+ candidates = CassandraQuery.getObjectsOfCandidateType(ec,
+ mconn, candidateClass, subclasses, ignoreCache,
+ fromInclNo, toExclNo);
+ }
+
+ // Apply any result restrictions to the results
+ JavaQueryEvaluator resultMapper = new JDOQLEvaluator(this,
+ candidates, compilation, parameters, ec
+ .getClassLoaderResolver());
+ Collection results = resultMapper.execute(true, true, true, true,
+ true);
+
+ if (NucleusLogger.QUERY.isDebugEnabled()) {
+ NucleusLogger.QUERY.debug(LOCALISER.msg("021074", "JDOQL", ""
+ + (System.currentTimeMillis() - startTime)));
+ }
+
+ return results;
+ } finally {
+ mconn.release();
+ }
+ }
}

0 comments on commit fa1c55a

Please sign in to comment.