Updated Branches: refs/heads/0.2-dev b809eac29 -> a2585810c
Working on https://issues.apache.org/jira/browse/BLUR-42. Not tested yet, and no cancel implementation. Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/a2585810 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/a2585810 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/a2585810 Branch: refs/heads/0.2-dev Commit: a2585810c720fa2559a941a07c9f9fde990d250c Parents: b809eac Author: Aaron McCurry <[email protected]> Authored: Tue Jan 29 09:57:59 2013 -0500 Committer: Aaron McCurry <[email protected]> Committed: Tue Jan 29 09:57:59 2013 -0500 ---------------------------------------------------------------------- .../java/org/apache/blur/server/BlurServer.java | 310 ++-- .../apache/blur/server/LocalSearchCallable.java | 88 + .../apache/blur/server/QueryStatusContainer.java | 71 + .../apache/blur/server/RemoteSearchCallable.java | 63 + .../org/apache/blur/server/SearchCallable.java | 84 +- .../java/org/apache/blur/server/SessionInfo.java | 5 + .../org/apache/blur/thrift/ThriftBlurServer.java | 21 +- .../org/apache/blur/thrift/generated/Analyzer.java | 36 +- .../apache/blur/thrift/generated/AnalyzerType.java | 36 +- .../org/apache/blur/thrift/generated/Blur.java | 1386 ++++++++++----- .../blur/thrift/generated/ClassDefinition.java | 44 +- .../org/apache/blur/thrift/generated/CpuTime.java | 524 ------ .../org/apache/blur/thrift/generated/Document.java | 36 +- .../apache/blur/thrift/generated/QueryArgs.java | 110 +- .../apache/blur/thrift/generated/QueryStatus.java | 686 ++++++-- .../org/apache/blur/thrift/generated/ScoreDoc.java | 32 +- .../apache/blur/thrift/generated/ShardLayout.java | 32 +- .../org/apache/blur/thrift/generated/Sort.java | 36 +- .../blur/thrift/generated/TableDescriptor.java | 44 +- .../apache/blur/thrift/generated/TableSchema.java | 36 +- .../apache/blur/thrift/generated/TopFieldDocs.java | 72 +- .../blur/thrift/generated/UpdatePackage.java | 36 +- .../apache/blur/jdbc/BlurResultSetDocuments.java | 14 +- .../src/main/scripts/interface/Blur.thrift | 102 +- .../src/main/scripts/interface/gen-html/Blur.html | 71 +- .../src/main/scripts/interface/gen-html/index.html | 1 - .../org/apache/blur/thrift/generated/Analyzer.java | 36 +- .../apache/blur/thrift/generated/AnalyzerType.java | 36 +- .../org/apache/blur/thrift/generated/Blur.java | 1386 ++++++++++----- .../blur/thrift/generated/ClassDefinition.java | 44 +- .../org/apache/blur/thrift/generated/CpuTime.java | 524 ------ .../org/apache/blur/thrift/generated/Document.java | 36 +- .../apache/blur/thrift/generated/QueryArgs.java | 110 +- .../apache/blur/thrift/generated/QueryStatus.java | 686 ++++++-- .../org/apache/blur/thrift/generated/ScoreDoc.java | 32 +- .../apache/blur/thrift/generated/ShardLayout.java | 32 +- .../org/apache/blur/thrift/generated/Sort.java | 36 +- .../blur/thrift/generated/TableDescriptor.java | 44 +- .../apache/blur/thrift/generated/TableSchema.java | 36 +- .../apache/blur/thrift/generated/TopFieldDocs.java | 72 +- .../blur/thrift/generated/UpdatePackage.java | 36 +- .../main/scripts/interface/gen-perl/Blur/Blur.pm | 528 ++++--- .../main/scripts/interface/gen-perl/Blur/Types.pm | 658 ++++---- .../src/main/scripts/interface/gen-rb/blur.rb | 50 +- .../main/scripts/interface/gen-rb/blur_types.rb | 97 +- 45 files changed, 4661 insertions(+), 3794 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a2585810/src/blur-core/src/main/java/org/apache/blur/server/BlurServer.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/org/apache/blur/server/BlurServer.java b/src/blur-core/src/main/java/org/apache/blur/server/BlurServer.java index 04bde9f..524b0b0 100644 --- a/src/blur-core/src/main/java/org/apache/blur/server/BlurServer.java +++ b/src/blur-core/src/main/java/org/apache/blur/server/BlurServer.java @@ -25,13 +25,13 @@ import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Map.Entry; import java.util.Random; import java.util.Set; import java.util.UUID; -import java.util.concurrent.Callable; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; @@ -48,10 +48,8 @@ import org.apache.blur.manager.writer.BlurIndex; import org.apache.blur.server.BlurServer.SearchAction.ACTION; import org.apache.blur.server.TableLayout.TYPE; import org.apache.blur.thrift.BException; -import org.apache.blur.thrift.BlurClientManager; +import org.apache.blur.thrift.BlurClient; import org.apache.blur.thrift.Connection; -import org.apache.blur.thrift.commands.BlurCommand; -import org.apache.blur.thrift.generated.Blur.Client; import org.apache.blur.thrift.generated.Blur.Iface; import org.apache.blur.thrift.generated.BlurException; import org.apache.blur.thrift.generated.Document; @@ -89,6 +87,32 @@ public class BlurServer extends TableAdmin implements Iface { private ExecutorService _searchExecutor; private int _dataFetchThreadCount = 32; private TableLayout _layout; + private Map<String, SessionInfo> sessions = new ConcurrentHashMap<String, SessionInfo>(); + private String _nodeName; + + static class SearchAction { + enum ACTION { + LOCAL, REMOTE + } + + ACTION type; + + SearchAction(int shardIndex, IndexSearcher indexSearcher) { + this.type = ACTION.LOCAL; + this.shardIndex = shardIndex; + this.indexSearcher = indexSearcher; + } + + SearchAction(int shardIndex, Connection remoteServer) { + this.type = ACTION.REMOTE; + this.shardIndex = shardIndex; + this.remoteServer = remoteServer; + } + + int shardIndex; + IndexSearcher indexSearcher; + Connection remoteServer; + } public void init() throws BlurException { _dataFetch = Executors.newThreadPool("data-fetch-", _dataFetchThreadCount); @@ -127,42 +151,8 @@ public class BlurServer extends TableAdmin implements Iface { } } - public long getMaxTimeToLive() { - return _maxTimeToLive; - } - - public void setMaxTimeToLive(long maxTimeToLive) { - _maxTimeToLive = maxTimeToLive; - } - - public int getMaxQueryCacheElements() { - return _maxQueryCacheElements; - } - - public void setMaxQueryCacheElements(int maxQueryCacheElements) { - _maxQueryCacheElements = maxQueryCacheElements; - } - - public void setIndexServer(IndexServer indexServer) { - _indexServer = indexServer; - } - - public int getDataFetchThreadCount() { - return _dataFetchThreadCount; - } - - public void setDataFetchThreadCount(int dataFetchThreadCount) { - _dataFetchThreadCount = dataFetchThreadCount; - } - - public void setConfiguration(BlurConfiguration conf) { - _configuration = conf; - } - // New interface from this point - private Map<String, SessionInfo> sessions = new ConcurrentHashMap<String, SessionInfo>(); - @Override public Session openReadSession(String table) throws BlurException, TException { String uuid = UUID.randomUUID().toString(); @@ -204,6 +194,8 @@ public class BlurServer extends TableAdmin implements Iface { @Override public List<TopFieldDocs> search(final Session session, final QueryArgs queryArgs) throws BlurException, TException { SessionInfo info = getSessionInfo(session); + QueryStatusContainer container = info.getQueryStatusContainer(); + container.start(queryArgs); try { Map<Integer, IndexSearcher> searchers = info.getSearchers(); List<Integer> shardIndexes = queryArgs.getShardIndexes(); @@ -222,32 +214,24 @@ public class BlurServer extends TableAdmin implements Iface { boolean doDocScores = queryArgs.isDoDocScores(); boolean doMaxScore = queryArgs.isDoMaxScore(); int numberToFetch = queryArgs.getNumberToFetch(); + for (SearchAction action : searchersToSearch) { + SearchCallable searchCallable; + Future<TopFieldDocs> future; final int shardIndex = action.shardIndex; if (action.type == ACTION.LOCAL) { - SearchCallable searchCallable = new SearchCallable(shardIndex, action.indexSearcher, after, query, filter, sort, numberToFetch, doDocScores, doMaxScore); - Future<TopFieldDocs> future = _indexSearcherExecutor.submit(searchCallable); - futures.add(future); + searchCallable = new LocalSearchCallable(shardIndex, action.indexSearcher, after, query, filter, sort, numberToFetch, doDocScores, doMaxScore); + future = _indexSearcherExecutor.submit(searchCallable); } else if (action.type == ACTION.REMOTE) { // @TODO need to send only one call per server, instead of one for // each shard in the table - final Connection connection = action.remoteServer; - Future<TopFieldDocs> future = _indexSearcherExecutor.submit(new Callable<TopFieldDocs>() { - @Override - public TopFieldDocs call() throws Exception { - List<TopFieldDocs> list = BlurClientManager.execute(connection, new BlurCommand<List<TopFieldDocs>>() { - @Override - public List<TopFieldDocs> call(Client client) throws BlurException, TException { - QueryArgs remoteQueryArgs = new QueryArgs(queryArgs); - remoteQueryArgs.addToShardIndexes(shardIndex); - return client.search(session, remoteQueryArgs); - } - }); - return list.iterator().next(); - } - }); - futures.add(future); + searchCallable = new RemoteSearchCallable(session, action.remoteServer, shardIndex, queryArgs); + future = _indexSearcherExecutor.submit(searchCallable); + } else { + throw new BlurException("Unsupported action of [" + action.type + "]", null); } + container.addSearchCallable(queryArgs, future, searchCallable); + futures.add(future); } List<TopFieldDocs> result = new ArrayList<TopFieldDocs>(futures.size()); @@ -258,33 +242,11 @@ public class BlurServer extends TableAdmin implements Iface { } catch (Throwable t) { LOG.error("Unknown error", t); throw new BException(t.getMessage(), t); + } finally { + container.end(queryArgs); } } - static class SearchAction { - enum ACTION { - LOCAL, REMOTE - } - - ACTION type; - - SearchAction(int shardIndex, IndexSearcher indexSearcher) { - this.type = ACTION.LOCAL; - this.shardIndex = shardIndex; - this.indexSearcher = indexSearcher; - } - - SearchAction(int shardIndex, Connection remoteServer) { - this.type = ACTION.REMOTE; - this.shardIndex = shardIndex; - this.remoteServer = remoteServer; - } - - int shardIndex; - IndexSearcher indexSearcher; - Connection remoteServer; - } - private Collection<SearchAction> getSearchActions(TableDescriptor tableDescriptor, List<Integer> shardIndexes, Map<Integer, IndexSearcher> searchers) throws BlurException { String name = tableDescriptor.getName(); int shardCount = tableDescriptor.getShardCount(); @@ -349,12 +311,8 @@ public class BlurServer extends TableAdmin implements Iface { // docLocations String table = session.getTableName(); Connection connection = getConnection(table, shardIndex); - return BlurClientManager.execute(connection, new BlurCommand<List<Document>>() { - @Override - public List<Document> call(Client client) throws BlurException, TException { - return client.doc(session, Arrays.asList(docLocation), fieldsToLoad); - } - }); + Iface client = BlurClient.getClient(connection); + return client.doc(session, Arrays.asList(docLocation), fieldsToLoad); } @Override @@ -421,13 +379,8 @@ public class BlurServer extends TableAdmin implements Iface { String table = generation.getTable(); int shardIndex = generation.getShardIndex(); Connection connection = getConnection(table, shardIndex); - BlurClientManager.execute(connection, new BlurCommand<Void>() { - @Override - public Void call(Client client) throws BlurException, TException { - client.blockUntilGenerationIsVisible(Arrays.asList(generation), forceRefresh); - return null; - } - }); + Iface client = BlurClient.getClient(connection); + client.blockUntilGenerationIsVisible(Arrays.asList(generation), forceRefresh); } @Override @@ -515,18 +468,89 @@ public class BlurServer extends TableAdmin implements Iface { } @Override - public void cancelQuery(Session session, long id) throws BlurException, TException { - throw new BlurException("Not implemented", null); + public void cancelQuery(final Session session, final String server, final String id) throws BlurException, TException { + SessionInfo sessionInfo = getSessionInfo(session); + if (server == null) { + // broadcast to all + List<String> serverList = serverList(); + for (String serverName : serverList) { + Iface client = BlurClient.getClient(serverName); + client.cancelQuery(session, serverName, id); + } + } else if (server.equals(_nodeName)) { + // execute locally + try { + QueryStatusContainer queryStatusContainer = sessionInfo.getQueryStatusContainer(); + queryStatusContainer.cancelQuery(id); + } catch (Exception e) { + LOG.error("Unknown error while trying to cancel query with id [{0}] for session [{1}].", e, id, sessionInfo); + throw new BException(e.getMessage(), e); + } + } else { + // route to correct server. + Iface client = BlurClient.getClient(server); + client.cancelQuery(session, server, id); + } } @Override - public List<Long> queryStatusIdList(Session session) throws BlurException, TException { - throw new BlurException("Not implemented", null); + public Map<String, List<String>> queryStatusIdList(Session session, String server) throws BlurException, TException { + SessionInfo sessionInfo = getSessionInfo(session); + if (server == null) { + // broadcast to all + Map<String, List<String>> result = new HashMap<String, List<String>>(); + List<String> serverList = serverList(); + for (String serverName : serverList) { + Iface client = BlurClient.getClient(serverName); + result.putAll(client.queryStatusIdList(session, serverName)); + } + return result; + } else if (server.equals(_nodeName)) { + // execute locally + try { + Map<String, List<String>> result = new HashMap<String, List<String>>(); + QueryStatusContainer queryStatusContainer = sessionInfo.getQueryStatusContainer(); + result.put(server, queryStatusContainer.getQueryIds()); + return result; + } catch (Exception e) { + LOG.error("Unknown error while trying to get query status id list for session [{0}].", e, sessionInfo); + throw new BException(e.getMessage(), e); + } + } else { + // route to correct server. + Iface client = BlurClient.getClient(server); + return client.queryStatusIdList(session, server); + } } @Override - public QueryStatus queryStatus(Session session, long id) throws BlurException, TException { - throw new BlurException("Not implemented", null); + public Map<String, QueryStatus> queryStatus(Session session, String server, String id) throws BlurException, TException { + SessionInfo sessionInfo = getSessionInfo(session); + if (server == null) { + // broadcast to all + Map<String, QueryStatus> result = new HashMap<String, QueryStatus>(); + List<String> serverList = serverList(); + for (String serverName : serverList) { + Iface client = BlurClient.getClient(serverName); + result.putAll(client.queryStatus(session, serverName, id)); + } + return result; + } else if (server.equals(_nodeName)) { + // execute locally + try { + Map<String, QueryStatus> result = new HashMap<String, QueryStatus>(); + QueryStatusContainer queryStatusContainer = sessionInfo.getQueryStatusContainer(); + result.put(server, queryStatusContainer.getQueryStatus(id)); + return result; + } catch (Exception e) { + LOG.error("Unknown error while trying to get query status with id [{0}] for session [{1}].", e, id, sessionInfo); + throw new BException(e.getMessage(), e); + } + } else { + // route to correct server. + Iface client = BlurClient.getClient(server); + return client.queryStatus(session, server, id); + } } @Override @@ -540,7 +564,7 @@ public class BlurServer extends TableAdmin implements Iface { TableSchema tableSchema = new TableSchema(); tableSchema.setCluster(_clusterStatus.getClusterName()); for (Entry<String, org.apache.blur.thrift.generated.TYPE> entry : typeMap.entrySet()) { - tableSchema.addToFields(new FieldSchema(entry.getKey(), entry.getValue())); + tableSchema.addToFields(new FieldSchema(entry.getKey(), entry.getValue())); } return tableSchema; } catch (IOException e) { @@ -549,36 +573,20 @@ public class BlurServer extends TableAdmin implements Iface { } } - public TableLayout getLayout() { - return _layout; - } - - public void setLayout(TableLayout layout) { - this._layout = layout; - } - private List<Generation> forwardAddDocuments(final MutateOptions options, final List<Document> documents) throws BlurException, TException, IOException { String table = options.getTable(); int shardIndex = options.getShardIndex(); Connection connection = getConnection(table, shardIndex); - return BlurClientManager.execute(connection, new BlurCommand<List<Generation>>() { - @Override - public List<Generation> call(Client client) throws BlurException, TException { - return client.addDocuments(options, documents); - } - }); + Iface client = BlurClient.getClient(connection); + return client.addDocuments(options, documents); } private List<Generation> forwardUpdateDocuments(final MutateOptions options, final List<UpdatePackage> updatePackages) throws BlurException, TException, IOException { String table = options.getTable(); int shardIndex = options.getShardIndex(); Connection connection = getConnection(table, shardIndex); - return BlurClientManager.execute(connection, new BlurCommand<List<Generation>>() { - @Override - public List<Generation> call(Client client) throws BlurException, TException { - return client.updateDocuments(options, updatePackages); - } - }); + Iface client = BlurClient.getClient(connection); + return client.updateDocuments(options, updatePackages); } private List<Generation> forwardDeleteDocumentsByQueries(final MutateOptions options, final List<org.apache.blur.thrift.generated.Query> queries) throws BlurException, @@ -586,24 +594,16 @@ public class BlurServer extends TableAdmin implements Iface { String table = options.getTable(); int shardIndex = options.getShardIndex(); Connection connection = getConnection(table, shardIndex); - return BlurClientManager.execute(connection, new BlurCommand<List<Generation>>() { - @Override - public List<Generation> call(Client client) throws BlurException, TException { - return client.deleteDocumentsByQueries(options, queries); - } - }); + Iface client = BlurClient.getClient(connection); + return client.deleteDocumentsByQueries(options, queries); } private List<Generation> forwardDeleteDocuments(final MutateOptions options, final List<Term> terms) throws BlurException, TException, IOException { String table = options.getTable(); int shardIndex = options.getShardIndex(); Connection connection = getConnection(table, shardIndex); - return BlurClientManager.execute(connection, new BlurCommand<List<Generation>>() { - @Override - public List<Generation> call(Client client) throws BlurException, TException { - return client.deleteDocuments(options, terms); - } - }); + Iface client = BlurClient.getClient(connection); + return client.deleteDocuments(options, terms); } private Connection getConnection(String table, int shardIndex) { @@ -616,4 +616,48 @@ public class BlurServer extends TableAdmin implements Iface { throw new BlurException("Not implemented", null); } + public void setNodeName(String nodeName) { + _nodeName = nodeName; + } + + public TableLayout getLayout() { + return _layout; + } + + public void setLayout(TableLayout layout) { + this._layout = layout; + } + + public long getMaxTimeToLive() { + return _maxTimeToLive; + } + + public void setMaxTimeToLive(long maxTimeToLive) { + _maxTimeToLive = maxTimeToLive; + } + + public int getMaxQueryCacheElements() { + return _maxQueryCacheElements; + } + + public void setMaxQueryCacheElements(int maxQueryCacheElements) { + _maxQueryCacheElements = maxQueryCacheElements; + } + + public void setIndexServer(IndexServer indexServer) { + _indexServer = indexServer; + } + + public int getDataFetchThreadCount() { + return _dataFetchThreadCount; + } + + public void setDataFetchThreadCount(int dataFetchThreadCount) { + _dataFetchThreadCount = dataFetchThreadCount; + } + + public void setConfiguration(BlurConfiguration conf) { + _configuration = conf; + } + } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a2585810/src/blur-core/src/main/java/org/apache/blur/server/LocalSearchCallable.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/org/apache/blur/server/LocalSearchCallable.java b/src/blur-core/src/main/java/org/apache/blur/server/LocalSearchCallable.java new file mode 100644 index 0000000..5ce7bfe --- /dev/null +++ b/src/blur-core/src/main/java/org/apache/blur/server/LocalSearchCallable.java @@ -0,0 +1,88 @@ +package org.apache.blur.server; + +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +import static org.apache.blur.utils.ThriftLuceneConversion.setShardIndexTopDocs; +import static org.apache.blur.utils.ThriftLuceneConversion.setShardIndexTopFieldDocs; +import static org.apache.blur.utils.ThriftLuceneConversion.toThrift; + +import java.io.IOException; + +import org.apache.blur.thrift.generated.BlurException; +import org.apache.blur.thrift.generated.TopFieldDocs; +import org.apache.lucene.search.Filter; +import org.apache.lucene.search.IndexSearcher; +import org.apache.lucene.search.Query; +import org.apache.lucene.search.ScoreDoc; +import org.apache.lucene.search.Sort; + +public class LocalSearchCallable extends SearchCallable { + + private final ScoreDoc after; + private final Sort sort; + private final Filter filter; + private final Query query; + private final IndexSearcher searcher; + private final int count; + private final boolean doDocScores; + private final boolean doMaxScore; + private final int shardIndex; + + LocalSearchCallable(int shardIndex, IndexSearcher searcher, ScoreDoc after, Query query, Filter filter, Sort sort, int count, boolean doDocScores, boolean doMaxScore) { + this.after = after; + this.searcher = searcher; + this.query = query; + this.filter = filter; + this.sort = sort; + this.count = count; + this.doDocScores = doDocScores; + this.doMaxScore = doMaxScore; + this.shardIndex = shardIndex; + } + + @Override + public TopFieldDocs call() throws Exception { + return addShardIndex(doSearch()); + } + + private TopFieldDocs addShardIndex(TopFieldDocs topFieldDocs) { + topFieldDocs.setShardIndex(shardIndex); + return topFieldDocs; + } + + private TopFieldDocs doSearch() throws IOException { + if (after == null) { + if (sort == null) { + return toThrift(setShardIndexTopDocs(shardIndex, searcher.search(query, filter, count))); + } else { + return toThrift(setShardIndexTopFieldDocs(shardIndex, searcher.search(query, filter, count, sort, doDocScores, doMaxScore))); + } + } else { + if (sort == null) { + return toThrift(setShardIndexTopDocs(shardIndex, searcher.searchAfter(after, query, filter, count))); + } else { + return toThrift(setShardIndexTopFieldDocs(shardIndex, + (org.apache.lucene.search.TopFieldDocs) searcher.searchAfter(after, query, filter, count, sort, doDocScores, doMaxScore))); + } + } + } + + @Override + public void cancelExecution() throws Exception { + throw new BlurException("No implementation yet.", null); + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a2585810/src/blur-core/src/main/java/org/apache/blur/server/QueryStatusContainer.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/org/apache/blur/server/QueryStatusContainer.java b/src/blur-core/src/main/java/org/apache/blur/server/QueryStatusContainer.java new file mode 100644 index 0000000..f0c0b5e --- /dev/null +++ b/src/blur-core/src/main/java/org/apache/blur/server/QueryStatusContainer.java @@ -0,0 +1,71 @@ +package org.apache.blur.server; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.Future; + +import org.apache.blur.thrift.generated.BlurException; +import org.apache.blur.thrift.generated.QueryArgs; +import org.apache.blur.thrift.generated.QueryState; +import org.apache.blur.thrift.generated.QueryStatus; +import org.apache.blur.thrift.generated.TopFieldDocs; + +public class QueryStatusContainer { + + static class QueryStatusInternal { + final QueryStatus queryStatus = new QueryStatus(); + QueryArgs queryArgs; + Thread thread; + } + + private Map<String, QueryStatusInternal> idLookup = new HashMap<String, QueryStatusInternal>(); + + public List<String> getQueryIds() { + return new ArrayList<String>(idLookup.keySet()); + } + + public QueryStatus getQueryStatus(String id) throws BlurException { + QueryStatusInternal queryStatusInternal = idLookup.get(id); + if (queryStatusInternal == null) { + throw new BlurException("Query id [" + id + "] not found in the current session.", null); + } + return queryStatusInternal.queryStatus; + } + + public void cancelQuery(String id) { + + } + + public void addSearchCallable(QueryArgs queryArgs, Future<TopFieldDocs> future, SearchCallable searchCallable) { + + } + + public void start(QueryArgs queryArgs) throws BlurException { + String id = queryArgs.getId(); + if (id == null) { + queryArgs.setId(UUID.randomUUID().toString()); + } + QueryStatusInternal queryStatusInternal = idLookup.get(id); + if (queryStatusInternal != null) { + throw new BlurException("Query id [" + id + "] already registered for the current session.", null); + } + queryStatusInternal = new QueryStatusInternal(); + queryStatusInternal.queryArgs = queryArgs; + queryStatusInternal.thread = Thread.currentThread(); + queryStatusInternal.queryStatus.startTime = System.currentTimeMillis(); + queryStatusInternal.queryStatus.state = QueryState.RUNNING; + } + + public void end(QueryArgs queryArgs) { + String id = queryArgs.getId(); + QueryStatusInternal queryStatusInternal = idLookup.get(id); + if (queryStatusInternal.queryStatus.state == QueryState.RUNNING) { + queryStatusInternal.queryStatus.state = QueryState.COMPLETE; + queryStatusInternal.queryStatus.endTime = System.currentTimeMillis(); + } + } + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a2585810/src/blur-core/src/main/java/org/apache/blur/server/RemoteSearchCallable.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/org/apache/blur/server/RemoteSearchCallable.java b/src/blur-core/src/main/java/org/apache/blur/server/RemoteSearchCallable.java new file mode 100644 index 0000000..c5713ac --- /dev/null +++ b/src/blur-core/src/main/java/org/apache/blur/server/RemoteSearchCallable.java @@ -0,0 +1,63 @@ +package org.apache.blur.server; + +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +import java.util.List; + +import org.apache.blur.thrift.BlurClientManager; +import org.apache.blur.thrift.Connection; +import org.apache.blur.thrift.commands.BlurCommand; +import org.apache.blur.thrift.generated.Blur.Client; +import org.apache.blur.thrift.generated.BlurException; +import org.apache.blur.thrift.generated.QueryArgs; +import org.apache.blur.thrift.generated.Session; +import org.apache.blur.thrift.generated.TopFieldDocs; +import org.apache.thrift.TException; + +public final class RemoteSearchCallable extends SearchCallable { + + private Connection connection; + private Session session; + private int shardIndex; + private QueryArgs queryArgs; + + public RemoteSearchCallable(Session session, Connection connection, int shardIndex, QueryArgs queryArgs) { + this.session = session; + this.connection = connection; + this.shardIndex = shardIndex; + this.queryArgs = queryArgs; + } + + @Override + public TopFieldDocs call() throws Exception { + List<TopFieldDocs> list = BlurClientManager.execute(connection, new BlurCommand<List<TopFieldDocs>>() { + @Override + public List<TopFieldDocs> call(Client client) throws BlurException, TException { + QueryArgs remoteQueryArgs = new QueryArgs(queryArgs); + remoteQueryArgs.addToShardIndexes(shardIndex); + return client.search(session, remoteQueryArgs); + } + }); + return list.iterator().next(); + } + + @Override + public void cancelExecution() throws Exception { + throw new BlurException("No implementation yet.", null); + } + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a2585810/src/blur-core/src/main/java/org/apache/blur/server/SearchCallable.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/org/apache/blur/server/SearchCallable.java b/src/blur-core/src/main/java/org/apache/blur/server/SearchCallable.java index 81b9158..5a3d209 100644 --- a/src/blur-core/src/main/java/org/apache/blur/server/SearchCallable.java +++ b/src/blur-core/src/main/java/org/apache/blur/server/SearchCallable.java @@ -1,66 +1,34 @@ package org.apache.blur.server; -import static org.apache.blur.utils.ThriftLuceneConversion.setShardIndexTopDocs; -import static org.apache.blur.utils.ThriftLuceneConversion.setShardIndexTopFieldDocs; -import static org.apache.blur.utils.ThriftLuceneConversion.toThrift; - -import java.io.IOException; +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ import java.util.concurrent.Callable; import org.apache.blur.thrift.generated.TopFieldDocs; -import org.apache.lucene.search.Filter; -import org.apache.lucene.search.IndexSearcher; -import org.apache.lucene.search.Query; -import org.apache.lucene.search.ScoreDoc; -import org.apache.lucene.search.Sort; - -public class SearchCallable implements Callable<TopFieldDocs> { - private final ScoreDoc after; - private final Sort sort; - private final Filter filter; - private final Query query; - private final IndexSearcher searcher; - private final int count; - private final boolean doDocScores; - private final boolean doMaxScore; - private final int shardIndex; - - SearchCallable(int shardIndex, IndexSearcher searcher, ScoreDoc after, Query query, Filter filter, Sort sort, int count, boolean doDocScores, boolean doMaxScore) { - this.after = after; - this.searcher = searcher; - this.query = query; - this.filter = filter; - this.sort = sort; - this.count = count; - this.doDocScores = doDocScores; - this.doMaxScore = doMaxScore; - this.shardIndex = shardIndex; - } - @Override - public TopFieldDocs call() throws Exception { - return addShardIndex(doSearch()); - } +public abstract class SearchCallable implements Callable<TopFieldDocs> { - private TopFieldDocs addShardIndex(TopFieldDocs topFieldDocs) { - topFieldDocs.setShardIndex(shardIndex); - return topFieldDocs; - } + /** + * This method will cancel the search being executed in another thread. + * Meaning this will only be called by a thread other than the main execution + * thread. + * + * @throws Exception + */ + public abstract void cancelExecution() throws Exception; - private TopFieldDocs doSearch() throws IOException { - if (after == null) { - if (sort == null) { - return toThrift(setShardIndexTopDocs(shardIndex, searcher.search(query, filter, count))); - } else { - return toThrift(setShardIndexTopFieldDocs(shardIndex, searcher.search(query, filter, count, sort, doDocScores, doMaxScore))); - } - } else { - if (sort == null) { - return toThrift(setShardIndexTopDocs(shardIndex, searcher.searchAfter(after, query, filter, count))); - } else { - return toThrift(setShardIndexTopFieldDocs(shardIndex, - (org.apache.lucene.search.TopFieldDocs) searcher.searchAfter(after, query, filter, count, sort, doDocScores, doMaxScore))); - } - } - } -} \ No newline at end of file +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a2585810/src/blur-core/src/main/java/org/apache/blur/server/SessionInfo.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/org/apache/blur/server/SessionInfo.java b/src/blur-core/src/main/java/org/apache/blur/server/SessionInfo.java index 3fcc9e9..be3109b 100644 --- a/src/blur-core/src/main/java/org/apache/blur/server/SessionInfo.java +++ b/src/blur-core/src/main/java/org/apache/blur/server/SessionInfo.java @@ -37,6 +37,7 @@ public class SessionInfo { private Map<Integer, IndexReader> readers = new HashMap<Integer, IndexReader>(); private Map<Integer, IndexSearcher> searchers = new HashMap<Integer, IndexSearcher>(); private TableDescriptor tableDescriptor; + private QueryStatusContainer queryStatusContainer = new QueryStatusContainer(); public BlurAnalyzer getAnalyzer() { return analyzer; @@ -97,4 +98,8 @@ public class SessionInfo { this.tableDescriptor = tableDescriptor; } + public QueryStatusContainer getQueryStatusContainer() { + return queryStatusContainer; + } + } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a2585810/src/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurServer.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurServer.java b/src/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurServer.java index 8e59dc5..23faf73 100644 --- a/src/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurServer.java +++ b/src/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurServer.java @@ -163,8 +163,6 @@ public class ThriftBlurServer extends AbstractThriftServer { BlurUtil.setupZookeeper(zooKeeper, configuration.get(BLUR_CLUSTER_NAME)); -// final SimpleZKClusterStatus clusterStatus = new SimpleZKClusterStatus(BlurConstants.BLUR_CLUSTER, zooKeeper); - final ZookeeperClusterStatus clusterStatus = new ZookeeperClusterStatus(BlurConstants.BLUR_CLUSTER, zooKeeper); final BlurIndexRefresher refresher = new BlurIndexRefresher(); @@ -213,15 +211,16 @@ public class ThriftBlurServer extends AbstractThriftServer { } }; - final BlurServer shardServer = new BlurServer(); - shardServer.setIndexServer(indexServer); - shardServer.setZookeeper(zooKeeper); - shardServer.setClusterStatus(clusterStatus); - shardServer.setConfiguration(configuration); - shardServer.setLayout(layout); - shardServer.init(); + final BlurServer blurServer = new BlurServer(); + blurServer.setIndexServer(indexServer); + blurServer.setZookeeper(zooKeeper); + blurServer.setClusterStatus(clusterStatus); + blurServer.setConfiguration(configuration); + blurServer.setLayout(layout); + blurServer.setNodeName(nodeName); + blurServer.init(); - Iface iface = BlurUtil.recordMethodCallsAndAverageTimes(blurMetrics, shardServer, Iface.class); + Iface iface = BlurUtil.recordMethodCallsAndAverageTimes(blurMetrics, blurServer, Iface.class); int threadCount = configuration.getInt(BLUR_SHARD_SERVER_THRIFT_THREAD_COUNT, 32); @@ -240,7 +239,7 @@ public class ThriftBlurServer extends AbstractThriftServer { @Override public void shutdown() { ThreadWatcher threadWatcher = ThreadWatcher.instance(); - quietClose(refresher, server, shardServer, indexServer, threadWatcher, clusterStatus, zooKeeper, httpServer); + quietClose(refresher, server, blurServer, indexServer, threadWatcher, clusterStatus, zooKeeper, httpServer); } }; server.setShutdown(shutdown); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a2585810/src/blur-thrift/src/main/java/org/apache/blur/thrift/generated/Analyzer.java ---------------------------------------------------------------------- diff --git a/src/blur-thrift/src/main/java/org/apache/blur/thrift/generated/Analyzer.java b/src/blur-thrift/src/main/java/org/apache/blur/thrift/generated/Analyzer.java index c009344..c5d4dea 100644 --- a/src/blur-thrift/src/main/java/org/apache/blur/thrift/generated/Analyzer.java +++ b/src/blur-thrift/src/main/java/org/apache/blur/thrift/generated/Analyzer.java @@ -376,14 +376,14 @@ public class Analyzer implements org.apache.thrift.TBase<Analyzer, Analyzer._Fie case 1: // ANALYZER_TYPES if (schemeField.type == org.apache.thrift.protocol.TType.LIST) { { - org.apache.thrift.protocol.TList _list36 = iprot.readListBegin(); - struct.analyzerTypes = new ArrayList<AnalyzerType>(_list36.size); - for (int _i37 = 0; _i37 < _list36.size; ++_i37) + org.apache.thrift.protocol.TList _list26 = iprot.readListBegin(); + struct.analyzerTypes = new ArrayList<AnalyzerType>(_list26.size); + for (int _i27 = 0; _i27 < _list26.size; ++_i27) { - AnalyzerType _elem38; // required - _elem38 = new AnalyzerType(); - _elem38.read(iprot); - struct.analyzerTypes.add(_elem38); + AnalyzerType _elem28; // required + _elem28 = new AnalyzerType(); + _elem28.read(iprot); + struct.analyzerTypes.add(_elem28); } iprot.readListEnd(); } @@ -411,9 +411,9 @@ public class Analyzer implements org.apache.thrift.TBase<Analyzer, Analyzer._Fie oprot.writeFieldBegin(ANALYZER_TYPES_FIELD_DESC); { oprot.writeListBegin(new org.apache.thrift.protocol.TList(org.apache.thrift.protocol.TType.STRUCT, struct.analyzerTypes.size())); - for (AnalyzerType _iter39 : struct.analyzerTypes) + for (AnalyzerType _iter29 : struct.analyzerTypes) { - _iter39.write(oprot); + _iter29.write(oprot); } oprot.writeListEnd(); } @@ -444,9 +444,9 @@ public class Analyzer implements org.apache.thrift.TBase<Analyzer, Analyzer._Fie if (struct.isSetAnalyzerTypes()) { { oprot.writeI32(struct.analyzerTypes.size()); - for (AnalyzerType _iter40 : struct.analyzerTypes) + for (AnalyzerType _iter30 : struct.analyzerTypes) { - _iter40.write(oprot); + _iter30.write(oprot); } } } @@ -458,14 +458,14 @@ public class Analyzer implements org.apache.thrift.TBase<Analyzer, Analyzer._Fie BitSet incoming = iprot.readBitSet(1); if (incoming.get(0)) { { - org.apache.thrift.protocol.TList _list41 = new org.apache.thrift.protocol.TList(org.apache.thrift.protocol.TType.STRUCT, iprot.readI32()); - struct.analyzerTypes = new ArrayList<AnalyzerType>(_list41.size); - for (int _i42 = 0; _i42 < _list41.size; ++_i42) + org.apache.thrift.protocol.TList _list31 = new org.apache.thrift.protocol.TList(org.apache.thrift.protocol.TType.STRUCT, iprot.readI32()); + struct.analyzerTypes = new ArrayList<AnalyzerType>(_list31.size); + for (int _i32 = 0; _i32 < _list31.size; ++_i32) { - AnalyzerType _elem43; // required - _elem43 = new AnalyzerType(); - _elem43.read(iprot); - struct.analyzerTypes.add(_elem43); + AnalyzerType _elem33; // required + _elem33 = new AnalyzerType(); + _elem33.read(iprot); + struct.analyzerTypes.add(_elem33); } } struct.setAnalyzerTypesIsSet(true); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a2585810/src/blur-thrift/src/main/java/org/apache/blur/thrift/generated/AnalyzerType.java ---------------------------------------------------------------------- diff --git a/src/blur-thrift/src/main/java/org/apache/blur/thrift/generated/AnalyzerType.java b/src/blur-thrift/src/main/java/org/apache/blur/thrift/generated/AnalyzerType.java index 58c36b3..b3e25bb 100644 --- a/src/blur-thrift/src/main/java/org/apache/blur/thrift/generated/AnalyzerType.java +++ b/src/blur-thrift/src/main/java/org/apache/blur/thrift/generated/AnalyzerType.java @@ -583,14 +583,14 @@ public class AnalyzerType implements org.apache.thrift.TBase<AnalyzerType, Analy case 3: // ANALYZER_SUB_TYPES if (schemeField.type == org.apache.thrift.protocol.TType.LIST) { { - org.apache.thrift.protocol.TList _list28 = iprot.readListBegin(); - struct.analyzerSubTypes = new ArrayList<AnalyzerSubType>(_list28.size); - for (int _i29 = 0; _i29 < _list28.size; ++_i29) + org.apache.thrift.protocol.TList _list18 = iprot.readListBegin(); + struct.analyzerSubTypes = new ArrayList<AnalyzerSubType>(_list18.size); + for (int _i19 = 0; _i19 < _list18.size; ++_i19) { - AnalyzerSubType _elem30; // required - _elem30 = new AnalyzerSubType(); - _elem30.read(iprot); - struct.analyzerSubTypes.add(_elem30); + AnalyzerSubType _elem20; // required + _elem20 = new AnalyzerSubType(); + _elem20.read(iprot); + struct.analyzerSubTypes.add(_elem20); } iprot.readListEnd(); } @@ -628,9 +628,9 @@ public class AnalyzerType implements org.apache.thrift.TBase<AnalyzerType, Analy oprot.writeFieldBegin(ANALYZER_SUB_TYPES_FIELD_DESC); { oprot.writeListBegin(new org.apache.thrift.protocol.TList(org.apache.thrift.protocol.TType.STRUCT, struct.analyzerSubTypes.size())); - for (AnalyzerSubType _iter31 : struct.analyzerSubTypes) + for (AnalyzerSubType _iter21 : struct.analyzerSubTypes) { - _iter31.write(oprot); + _iter21.write(oprot); } oprot.writeListEnd(); } @@ -673,9 +673,9 @@ public class AnalyzerType implements org.apache.thrift.TBase<AnalyzerType, Analy if (struct.isSetAnalyzerSubTypes()) { { oprot.writeI32(struct.analyzerSubTypes.size()); - for (AnalyzerSubType _iter32 : struct.analyzerSubTypes) + for (AnalyzerSubType _iter22 : struct.analyzerSubTypes) { - _iter32.write(oprot); + _iter22.write(oprot); } } } @@ -696,14 +696,14 @@ public class AnalyzerType implements org.apache.thrift.TBase<AnalyzerType, Analy } if (incoming.get(2)) { { - org.apache.thrift.protocol.TList _list33 = new org.apache.thrift.protocol.TList(org.apache.thrift.protocol.TType.STRUCT, iprot.readI32()); - struct.analyzerSubTypes = new ArrayList<AnalyzerSubType>(_list33.size); - for (int _i34 = 0; _i34 < _list33.size; ++_i34) + org.apache.thrift.protocol.TList _list23 = new org.apache.thrift.protocol.TList(org.apache.thrift.protocol.TType.STRUCT, iprot.readI32()); + struct.analyzerSubTypes = new ArrayList<AnalyzerSubType>(_list23.size); + for (int _i24 = 0; _i24 < _list23.size; ++_i24) { - AnalyzerSubType _elem35; // required - _elem35 = new AnalyzerSubType(); - _elem35.read(iprot); - struct.analyzerSubTypes.add(_elem35); + AnalyzerSubType _elem25; // required + _elem25 = new AnalyzerSubType(); + _elem25.read(iprot); + struct.analyzerSubTypes.add(_elem25); } } struct.setAnalyzerSubTypesIsSet(true);
