You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by yo...@apache.org on 2012/01/06 22:08:39 UTC
svn commit: r1228407 - in
/lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr:
handler/component/RealTimeGetComponent.java update/UpdateLog.java
Author: yonik
Date: Fri Jan 6 21:08:39 2012
New Revision: 1228407
URL: http://svn.apache.org/viewvc?rev=1228407&view=rev
Log:
add retrieval of tlog records by version
Modified:
lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/handler/component/RealTimeGetComponent.java
lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/update/UpdateLog.java
Modified: lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/handler/component/RealTimeGetComponent.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/handler/component/RealTimeGetComponent.java?rev=1228407&r1=1228406&r2=1228407&view=diff
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/handler/component/RealTimeGetComponent.java (original)
+++ lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/handler/component/RealTimeGetComponent.java Fri Jan 6 21:08:39 2012
@@ -40,6 +40,8 @@ import org.apache.solr.search.SolrIndexS
import org.apache.solr.update.DocumentBuilder;
import org.apache.solr.update.UpdateLog;
import org.apache.solr.util.RefCounted;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import javax.xml.transform.Transformer;
import java.io.IOException;
@@ -55,6 +57,7 @@ import java.util.List;
*/
public class RealTimeGetComponent extends SearchComponent
{
+ public static Logger log = LoggerFactory.getLogger(UpdateLog.class);
public static final String COMPONENT_NAME = "get";
@Override
@@ -81,8 +84,13 @@ public class RealTimeGetComponent extend
processGetVersions(rb);
return;
}
-
-
+
+ val = params.get("getUpdates");
+ if (val != null) {
+ processGetUpdates(rb);
+ return;
+ }
+
String id[] = params.getParams("id");
String ids[] = params.getParams("ids");
@@ -287,6 +295,51 @@ public class RealTimeGetComponent extend
}
+ public void processGetUpdates(ResponseBuilder rb) throws IOException
+ {
+ SolrQueryRequest req = rb.req;
+ SolrQueryResponse rsp = rb.rsp;
+ SolrParams params = req.getParams();
+
+ if (!params.getBool(COMPONENT_NAME, true)) {
+ return;
+ }
+ String versionsStr = params.get("getUpdates");
+ if (versionsStr == null) return;
+
+ UpdateLog ulog = req.getCore().getUpdateHandler().getUpdateLog();
+ if (ulog == null) return;
+
+ List<String> versions = StrUtils.splitSmart(versionsStr, ",", true);
+
+ // TODO: get this from cache instead of rebuilding?
+ UpdateLog.RecentUpdates recentUpdates = ulog.getRecentUpdates();
+
+ List<Object> updates = new ArrayList<Object>(versions.size());
+
+ try {
+ for (String versionStr : versions) {
+ long version = Long.parseLong(versionStr);
+ try {
+ Object o = recentUpdates.lookup(version);
+ if (o == null) continue;
+
+ // TODO: do any kind of validation here?
+ updates.add(o);
+
+ } catch (SolrException e) {
+ log.warn("Exception reverse reading log", e);
+ } catch (ClassCastException e) {
+ log.warn("Exception reverse reading log", e);
+ }
+ }
+
+ rb.rsp.add("updates", updates);
+
+ } finally {
+ recentUpdates.close(); // cache this somehow?
+ }
+ }
}
Modified: lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/update/UpdateLog.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/update/UpdateLog.java?rev=1228407&r1=1228406&r2=1228407&view=diff
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/update/UpdateLog.java (original)
+++ lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/update/UpdateLog.java Fri Jan 6 21:08:39 2012
@@ -560,16 +560,24 @@ public class UpdateLog implements Plugin
}
+ static class Update {
+ TransactionLog log;
+ long version;
+ long pointer;
+ }
+
// TODO: nocommit: decrement the references of the transaction logs at some point
public class RecentUpdates {
Deque<TransactionLog> logList; // newest first
- List<List<LogPtr>> updates;
+ List<List<Update>> updateList;
+ HashMap<Long, Update> updates;
+
public List<Long> getVersions(int n) {
List<Long> ret = new ArrayList(n);
- for (List<LogPtr> singleList : updates) {
- for (LogPtr ptr : singleList) {
+ for (List<Update> singleList : updateList) {
+ for (Update ptr : singleList) {
ret.add(ptr.version);
if (--n <= 0) return ret;
}
@@ -577,13 +585,21 @@ public class UpdateLog implements Plugin
return ret;
}
+
+ public Object lookup(long version) {
+ Update update = updates.get(version);
+ if (update == null) return null;
+
+ return update.log.lookup(update.pointer);
+ }
private void update() {
int numUpdates = 0;
- updates = new ArrayList<List<LogPtr>>(logList.size());
+ updateList = new ArrayList<List<Update>>(logList.size());
+ updates = new HashMap<Long,Update>(numRecordsToKeep);
for (TransactionLog oldLog : logList) {
- List<LogPtr> updatesForLog = new ArrayList<LogPtr>();
+ List<Update> updatesForLog = new ArrayList<Update>();
TransactionLog.ReverseReader reader = null;
try {
@@ -605,8 +621,13 @@ public class UpdateLog implements Plugin
case UpdateLog.ADD:
case UpdateLog.DELETE:
case UpdateLog.DELETE_BY_QUERY:
- LogPtr ptr = new LogPtr(reader.position(), version);
- updatesForLog.add(ptr);
+ Update update = new Update();
+ update.log = oldLog;
+ update.pointer = reader.position();
+ update.version = version;
+
+ updatesForLog.add(update);
+ updates.put(version, update);
break;
case UpdateLog.COMMIT:
@@ -630,7 +651,7 @@ public class UpdateLog implements Plugin
if (reader != null) reader.close();
}
- updates.add(updatesForLog);
+ updateList.add(updatesForLog);
}
}