Removed the session parameter from a few methods in the interface. Now the query statuses are tracked at the server level instead of inside each session.
Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/cf92e0ff Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/cf92e0ff Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/cf92e0ff Branch: refs/heads/0.2-dev Commit: cf92e0ff8749b3d005bdd7e562e439a0149ee700 Parents: cb2e5c5 Author: Aaron McCurry <[email protected]> Authored: Sat Mar 2 18:33:07 2013 -0500 Committer: Aaron McCurry <[email protected]> Committed: Sat Mar 2 18:33:07 2013 -0500 ---------------------------------------------------------------------- .../manager/clusterstatus/BaseClusterStatus.java | 2 +- .../blur/manager/clusterstatus/ClusterStatus.java | 2 +- .../clusterstatus/ZookeeperClusterStatus.java | 2 +- .../java/org/apache/blur/server/BlurServer.java | 59 +- .../apache/blur/server/QueryStatusContainer.java | 35 +- .../java/org/apache/blur/server/SessionInfo.java | 7 +- .../java/org/apache/blur/server/TableAdmin.java | 93 ++- .../java/org/apache/blur/server/TableContext.java | 81 +-- .../org/apache/blur/thrift/ThriftBlurServer.java | 1 + .../java/org/apache/blur/utils/BlurConstants.java | 1 + .../src/main/java/org/apache/blur/shell/Main.java | 9 +- .../org/apache/blur/shell/SchemaTableCommand.java | 5 +- .../org/apache/blur/thrift/generated/Blur.java | 596 ++++----------- .../src/main/scripts/interface/Blur.thrift | 15 +- .../src/main/scripts/interface/gen-html/Blur.html | 18 +- .../org/apache/blur/thrift/generated/Blur.java | 596 ++++----------- .../main/scripts/interface/gen-perl/Blur/Blur.pm | 142 +--- .../src/main/scripts/interface/gen-rb/blur.rb | 60 +- 18 files changed, 483 insertions(+), 1241 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/cf92e0ff/src/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/BaseClusterStatus.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/BaseClusterStatus.java b/src/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/BaseClusterStatus.java index e7a7e5b..13c2e35 100644 --- a/src/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/BaseClusterStatus.java +++ b/src/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/BaseClusterStatus.java @@ -129,7 +129,7 @@ public abstract class BaseClusterStatus extends ClusterStatus { } @Override - public void removeTable(String table, boolean deleteIndexFiles) { + public void removeTable(String table) { try { String tablePath = ZookeeperPathConstants.getTablePath(getClusterName(), table); if (getZooKeeper().exists(tablePath, false) == null) { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/cf92e0ff/src/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/ClusterStatus.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/ClusterStatus.java b/src/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/ClusterStatus.java index 40c9e8f..dc08474 100644 --- a/src/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/ClusterStatus.java +++ b/src/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/ClusterStatus.java @@ -50,7 +50,7 @@ public abstract class ClusterStatus { public abstract void enableTable(String table); - public abstract void removeTable(String table, boolean deleteIndexFiles); + public abstract void removeTable(String table); public abstract boolean isOpen(); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/cf92e0ff/src/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/ZookeeperClusterStatus.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/ZookeeperClusterStatus.java b/src/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/ZookeeperClusterStatus.java index b08d8fa..52419a8 100644 --- a/src/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/ZookeeperClusterStatus.java +++ b/src/blur-core/src/main/java/org/apache/blur/manager/clusterstatus/ZookeeperClusterStatus.java @@ -410,7 +410,7 @@ public class ZookeeperClusterStatus extends ClusterStatus { } @Override - public void removeTable(String table, boolean deleteIndexFiles) { + public void removeTable(String table) { long s = System.nanoTime(); try { checkIfOpen(); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/cf92e0ff/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 8697ae3..0157a25 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 @@ -16,6 +16,7 @@ package org.apache.blur.server; * See the License for the specific language governing permissions and * limitations under the License. */ +import static org.apache.blur.utils.BlurConstants.BLUR_QUERYSTATUS_TRACKING_MAX; import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_CACHE_MAX_QUERYCACHE_ELEMENTS; import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_CACHE_MAX_TIMETOLIVE; import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_DATA_FETCH_THREAD_COUNT; @@ -89,7 +90,8 @@ public class BlurServer extends TableAdmin implements Iface { private TableLayout _layout; private Map<String, SessionInfo> sessions = new ConcurrentHashMap<String, SessionInfo>(); private String _nodeName; - + private QueryStatusContainer _queryStatusContainer; + static class SearchAction { enum ACTION { LOCAL, REMOTE @@ -125,6 +127,7 @@ public class BlurServer extends TableAdmin implements Iface { _dataFetchThreadCount = _configuration.getInt(BLUR_SHARD_DATA_FETCH_THREAD_COUNT, 8); _maxQueryCacheElements = _configuration.getInt(BLUR_SHARD_CACHE_MAX_QUERYCACHE_ELEMENTS, 128); _maxTimeToLive = _configuration.getLong(BLUR_SHARD_CACHE_MAX_TIMETOLIVE, TimeUnit.MINUTES.toMillis(1)); + _queryStatusContainer = new QueryStatusContainer(_configuration.getInt(BLUR_QUERYSTATUS_TRACKING_MAX, 16384)); } @Override @@ -194,8 +197,7 @@ 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); + String queryId = _queryStatusContainer.start(queryArgs); try { Map<Integer, IndexSearcher> searchers = info.getSearchers(); List<Integer> shardIndexes = queryArgs.getShardIndexes(); @@ -230,7 +232,7 @@ public class BlurServer extends TableAdmin implements Iface { } else { throw new BlurException("Unsupported action of [" + action.type + "]", null); } - container.addSearchCallable(queryArgs, future, searchCallable); + _queryStatusContainer.addSearchCallable(queryArgs, future, searchCallable); futures.add(future); } @@ -239,15 +241,15 @@ public class BlurServer extends TableAdmin implements Iface { result.add(future.get()); } if (queryArgs.isAggregateResults()) { - return aggregateResults(result,numberToFetch); + return aggregateResults(result, numberToFetch); } else { - return result; + return result; } } catch (Throwable t) { LOG.error("Unknown error", t); throw new BException(t.getMessage(), t); } finally { - container.end(queryArgs); + _queryStatusContainer.end(queryId); } } @@ -481,96 +483,91 @@ public class BlurServer extends TableAdmin implements Iface { } @Override - public void cancelQuery(final Session session, final String server, final String id) throws BlurException, TException { - SessionInfo sessionInfo = getSessionInfo(session); + public void cancelQuery(final String server, final String id) throws BlurException, TException { if (server == null) { // broadcast to all List<String> serverList = serverList(); for (String serverName : serverList) { Iface client = BlurClient.getClient(serverName); - client.cancelQuery(session, serverName, id); + client.cancelQuery(serverName, id); } } else if (server.equals(_nodeName)) { // execute locally try { - QueryStatusContainer queryStatusContainer = sessionInfo.getQueryStatusContainer(); - queryStatusContainer.cancelQuery(id); + _queryStatusContainer.cancelQuery(id); } catch (Exception e) { - LOG.error("Unknown error while trying to cancel query with id [{0}] for session [{1}].", e, id, sessionInfo); + LOG.error("Unknown error while trying to cancel query with id [{0}].", e, id); throw new BException(e.getMessage(), e); } } else { // route to correct server. Iface client = BlurClient.getClient(server); - client.cancelQuery(session, server, id); + client.cancelQuery(server, id); } } @Override - public Map<String, List<String>> queryStatusIdList(Session session, String server) throws BlurException, TException { - SessionInfo sessionInfo = getSessionInfo(session); + public Map<String, List<String>> queryStatusIdList(String server) throws BlurException, TException { 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)); + result.putAll(client.queryStatusIdList(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()); + 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); + LOG.error("Unknown error while trying to get query status id list.", e); throw new BException(e.getMessage(), e); } } else { // route to correct server. Iface client = BlurClient.getClient(server); - return client.queryStatusIdList(session, server); + return client.queryStatusIdList(server); } } @Override - public Map<String, QueryStatus> queryStatus(Session session, String server, String id) throws BlurException, TException { - SessionInfo sessionInfo = getSessionInfo(session); + public Map<String, QueryStatus> queryStatus(String server, String id) throws BlurException, TException { 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)); + result.putAll(client.queryStatus(serverName, id)); } return result; } else if (server.equals(_nodeName)) { // execute locally try { Map<String, QueryStatus> result = new HashMap<String, QueryStatus>(); - QueryStatusContainer queryStatusContainer = sessionInfo.getQueryStatusContainer(); - QueryStatus queryStatus = queryStatusContainer.getQueryStatus(id); + QueryStatus queryStatus = _queryStatusContainer.getQueryStatus(id); if (queryStatus != null) { - result.put(server, queryStatus); + result.put(server, queryStatus); } return result; } catch (Exception e) { - LOG.error("Unknown error while trying to get query status with id [{0}] for session [{1}].", e, id, sessionInfo); + LOG.error("Unknown error while trying to get query status with id [{0}].", e, id); throw new BException(e.getMessage(), e); } } else { // route to correct server. Iface client = BlurClient.getClient(server); - return client.queryStatus(session, server, id); + return client.queryStatus(server, id); } } @Override - public TableSchema schema(Session session) throws BlurException, TException { + public TableSchema schema(String tableName) throws BlurException, TException { + Session session = newSession(tableName, UUID.randomUUID().toString()); SessionInfo info = getSessionInfo(session); String name = info.getTableDescriptor().getName(); try { @@ -586,6 +583,8 @@ public class BlurServer extends TableAdmin implements Iface { } catch (IOException e) { LOG.error("Unknown error while trying to get the schema for the table [{0}].", e, name); throw new BException(e.getMessage(), e); + } finally { + closeReadSession(session); } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/cf92e0ff/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 index 858d707..93894f8 100644 --- 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 @@ -17,9 +17,7 @@ package org.apache.blur.server; * limitations under the License. */ 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; @@ -31,8 +29,11 @@ import org.apache.blur.thrift.generated.QueryState; import org.apache.blur.thrift.generated.QueryStatus; import org.apache.blur.thrift.generated.TopFieldDocs; +import com.googlecode.concurrentlinkedhashmap.ConcurrentLinkedHashMap; +import com.googlecode.concurrentlinkedhashmap.EvictionListener; + public class QueryStatusContainer { - + private static final Log LOG = LogFactory.getLog(QueryStatusContainer.class); static class QueryStatusInternal { @@ -41,7 +42,20 @@ public class QueryStatusContainer { Thread thread; } - private Map<String, QueryStatusInternal> idLookup = new HashMap<String, QueryStatusInternal>(); + private ConcurrentLinkedHashMap<String, QueryStatusInternal> idLookup; + + public QueryStatusContainer(long maxEntries) { + EvictionListener<String, QueryStatusInternal> listener = new EvictionListener<String, QueryStatusInternal>() { + @Override + public void onEviction(String key, QueryStatusInternal status) { + QueryStatus queryStatus = status.queryStatus; + if (queryStatus.state == QueryState.RUNNING) { + LOG.warn("Tracking too many queries, Query [{0}] evicted while still RUNNING", queryStatus); + } + } + }; + idLookup = new ConcurrentLinkedHashMap.Builder<String, QueryStatusInternal>().maximumWeightedCapacity(maxEntries).listener(listener).build(); + } public List<String> getQueryIds() { return new ArrayList<String>(idLookup.keySet()); @@ -63,26 +77,27 @@ public class QueryStatusContainer { } - public void start(QueryArgs queryArgs) throws BlurException { + public String start(QueryArgs queryArgs) throws BlurException { String id = queryArgs.getId(); if (id == null) { - queryArgs.setId(UUID.randomUUID().toString()); + id = UUID.randomUUID().toString(); + queryArgs.setId(id); } QueryStatusInternal queryStatusInternal = idLookup.get(id); if (queryStatusInternal != null) { - LOG.info("Query id [{0}] already registered for the current session.",id); + LOG.info("Query id [{0}] already registered for the current session.", id); } queryStatusInternal = new QueryStatusInternal(); queryStatusInternal.queryArgs = queryArgs; queryStatusInternal.thread = Thread.currentThread(); queryStatusInternal.queryStatus.startTime = System.currentTimeMillis(); queryStatusInternal.queryStatus.state = QueryState.RUNNING; - + idLookup.put(queryArgs.getId(), queryStatusInternal); + return queryArgs.getId(); } - public void end(QueryArgs queryArgs) { - String id = queryArgs.getId(); + public void end(String id) { QueryStatusInternal queryStatusInternal = idLookup.get(id); if (queryStatusInternal.queryStatus.state == QueryState.RUNNING) { queryStatusInternal.queryStatus.state = QueryState.COMPLETE; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/cf92e0ff/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 be3109b..b2f4522 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,7 +37,6 @@ 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; @@ -83,7 +82,9 @@ public class SessionInfo { for (Entry<Integer, IndexReader> entry : readers.entrySet()) { IndexReader reader = entry.getValue(); try { + LOG.info("Before release reader ref count [{0}] with reader [{1}]", reader.getRefCount(), reader); reader.decRef(); + LOG.info("After release reader ref count [{0}] with reader [{1}]", reader.getRefCount(), reader); } catch (IOException e) { LOG.error("Unknown exception while trying to decRef on reader [{0}]", e, reader); } @@ -98,8 +99,4 @@ public class SessionInfo { this.tableDescriptor = tableDescriptor; } - public QueryStatusContainer getQueryStatusContainer() { - return queryStatusContainer; - } - } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/cf92e0ff/src/blur-core/src/main/java/org/apache/blur/server/TableAdmin.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/org/apache/blur/server/TableAdmin.java b/src/blur-core/src/main/java/org/apache/blur/server/TableAdmin.java index 813a9f6..265e1e5 100644 --- a/src/blur-core/src/main/java/org/apache/blur/server/TableAdmin.java +++ b/src/blur-core/src/main/java/org/apache/blur/server/TableAdmin.java @@ -16,6 +16,7 @@ package org.apache.blur.server; * See the License for the specific language governing permissions and * limitations under the License. */ +import java.io.IOException; import java.util.List; import java.util.Map; @@ -27,10 +28,12 @@ import org.apache.blur.thrift.BException; import org.apache.blur.thrift.generated.Blur.Iface; import org.apache.blur.thrift.generated.BlurException; import org.apache.blur.thrift.generated.TableDescriptor; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; import org.apache.thrift.TException; import org.apache.zookeeper.ZooKeeper; - public abstract class TableAdmin implements Iface { private static final Log LOG = LogFactory.getLog(TableAdmin.class); @@ -38,6 +41,7 @@ public abstract class TableAdmin implements Iface { protected ClusterStatus _clusterStatus; protected BlurConfiguration _configuration; private String cluster; + protected Configuration _conf; @Override public boolean isInSafeMode() throws BlurException, TException { @@ -52,6 +56,7 @@ public abstract class TableAdmin implements Iface { @Override public final void createTable(TableDescriptor tableDescriptor) throws BlurException, TException { try { + TableContext.clear(); _clusterStatus.createTable(tableDescriptor); } catch (Exception e) { LOG.error("Unknown error during create of [table={0}, tableDescriptor={1}]", e, tableDescriptor.name, tableDescriptor); @@ -65,17 +70,20 @@ public abstract class TableAdmin implements Iface { @Override public final void disableTable(String table) throws BlurException, TException { try { + TableContext.clear(); + TableDescriptor tableDescriptor = _clusterStatus.getTableDescriptor(true, table); _clusterStatus.disableTable(table); waitForTheTableToDisable(table); - waitForTheTableToDisengage(table); + waitForTheTableToDisengage(table, tableDescriptor); } catch (Exception e) { LOG.error("Unknown error during disable of [table={0}]", e, table); throw new BException(e.getMessage(), e); } } - private void waitForTheTableToDisengage(String table) throws BlurException, TException { - // LOG.info("Waiting for shards to disengage on table [" + table + "]"); + private void waitForTheTableToDisengage(String table, TableDescriptor tableDescriptor) throws BlurException, TException { + LOG.info("Waiting for shards to disengage on table [" + table + "]"); + } private void waitForTheTableToDisable(String table) throws BlurException, TException { @@ -97,6 +105,7 @@ public abstract class TableAdmin implements Iface { @Override public final void enableTable(String table) throws BlurException, TException { try { + TableContext.clear(); _clusterStatus.enableTable(table); waitForTheTableToEnable(table); waitForTheTableToEngage(table); @@ -124,29 +133,33 @@ public abstract class TableAdmin implements Iface { private void waitForTheTableToEngage(String table) throws BlurException, TException { LOG.info("IMPLEMENT - Waiting for shards to engage on table [" + table + "]"); - -// TableDescriptor tableDescriptor = _clusterStatus.getTableDescriptor(false, table); -// int shardCount = tableDescriptor.shardCount; -// LOG.info("Waiting for shards to engage on table [" + table + "]"); -// while (true) { -// try { -// Thread.sleep(3000); -// } catch (InterruptedException e) { -// LOG.error("Unknown error while engaging table [" + table + "]", e); -// throw new BException("Unknown error while engaging table [" + table + "]", e); -// } -// try { -// Map<String, String> shardServerLayout = getLayout();//shardServerLayout(table); -// LOG.info("Shards [" + shardServerLayout.size() + "/" + shardCount + "] of table [" + table + "] engaged"); -// if (shardServerLayout.size() == shardCount) { -// return; -// } -// } catch (BlurException e) { -// LOG.info("Stilling waiting", e); -// } catch (TException e) { -// LOG.info("Stilling waiting", e); -// } -// } + + // TableDescriptor tableDescriptor = + // _clusterStatus.getTableDescriptor(false, table); + // int shardCount = tableDescriptor.shardCount; + // LOG.info("Waiting for shards to engage on table [" + table + "]"); + // while (true) { + // try { + // Thread.sleep(3000); + // } catch (InterruptedException e) { + // LOG.error("Unknown error while engaging table [" + table + "]", e); + // throw new BException("Unknown error while engaging table [" + table + + // "]", e); + // } + // try { + // Map<String, String> shardServerLayout = + // getLayout();//shardServerLayout(table); + // LOG.info("Shards [" + shardServerLayout.size() + "/" + shardCount + + // "] of table [" + table + "] engaged"); + // if (shardServerLayout.size() == shardCount) { + // return; + // } + // } catch (BlurException e) { + // LOG.info("Stilling waiting", e); + // } catch (TException e) { + // LOG.info("Stilling waiting", e); + // } + // } } private Map<String, String> getLayout() throws BlurException, TException { @@ -156,13 +169,25 @@ public abstract class TableAdmin implements Iface { @Override public final void removeTable(String table, boolean deleteIndexFiles) throws BlurException, TException { try { - _clusterStatus.removeTable(table, deleteIndexFiles); + TableContext.clear(); + TableDescriptor tableDescriptor = _clusterStatus.getTableDescriptor(true, table); + _clusterStatus.removeTable(table); + if (deleteIndexFiles) { + removeIndexFiles(tableDescriptor); + } } catch (Exception e) { LOG.error("Unknown error during remove of [table={0}]", e, table); throw new BException(e.getMessage(), e); } } + private void removeIndexFiles(TableDescriptor tableDescriptor) throws IOException { + String storagePath = tableDescriptor.getStoragePath(); + Path path = new Path(storagePath); + FileSystem fileSystem = path.getFileSystem(_conf); + fileSystem.delete(path, true); + } + public void checkTable(String table) throws BlurException { if (table == null) { throw new BlurException("Table cannot be null.", null); @@ -183,7 +208,6 @@ public abstract class TableAdmin implements Iface { } } - public void checkForUpdates(String table) throws BlurException { TableDescriptor tableDescriptor = _clusterStatus.getTableDescriptor(true, table); if (tableDescriptor.isReadOnly()) { @@ -191,7 +215,6 @@ public abstract class TableAdmin implements Iface { } } - @Override public final List<String> serverList() throws BlurException, TException { try { @@ -233,8 +256,16 @@ public abstract class TableAdmin implements Iface { public void setZookeeper(ZooKeeper zookeeper) { _zookeeper = zookeeper; } - + public void setConfiguration(BlurConfiguration config) { _configuration = config; } + + public Configuration getConf() { + return _conf; + } + + public void setConf(Configuration conf) { + _conf = conf; + } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/cf92e0ff/src/blur-core/src/main/java/org/apache/blur/server/TableContext.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/org/apache/blur/server/TableContext.java b/src/blur-core/src/main/java/org/apache/blur/server/TableContext.java index a7e2ed4..21beba4 100644 --- a/src/blur-core/src/main/java/org/apache/blur/server/TableContext.java +++ b/src/blur-core/src/main/java/org/apache/blur/server/TableContext.java @@ -24,6 +24,7 @@ import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_TIME_BETWEEN_REFRES import java.util.Map; import java.util.Map.Entry; +import java.util.concurrent.ConcurrentHashMap; import org.apache.blur.analysis.BlurAnalyzer; import org.apache.blur.log.Log; @@ -58,20 +59,24 @@ public class TableContext { private long timeBetweenCommits; private long timeBetweenRefreshs; private TypeManager typeChecker; - private ScoreType defaultScoreType; - private Term defaultPrimeDocTerm; - public TypeManager getTypeManager() { - return typeChecker; - } + private static ConcurrentHashMap<String, TableContext> cache = new ConcurrentHashMap<String, TableContext>(); protected TableContext() { } + + public static void clear() { + cache.clear(); + } public static TableContext create(TableDescriptor tableDescriptor, TypeManager typeChecker) { + TableContext tableContext = cache.get(tableDescriptor.getName()); + if (tableContext != null) { + return tableContext; + } LOG.info("Creating table context for table [{0}]", tableDescriptor.getName()); Configuration configuration = new Configuration(); Map<String, String> properties = tableDescriptor.getProperties(); @@ -81,7 +86,7 @@ public class TableContext { } } - TableContext tableContext = new TableContext(); + tableContext = new TableContext(); tableContext.configuration = configuration; tableContext.tablePath = new Path(tableDescriptor.getStoragePath()); tableContext.walTablePath = new Path(tableContext.tablePath, LOGS); @@ -92,7 +97,7 @@ public class TableContext { tableContext.timeBetweenCommits = configuration.getLong(BLUR_SHARD_TIME_BETWEEN_COMMITS, 60000); tableContext.timeBetweenRefreshs = configuration.getLong(BLUR_SHARD_TIME_BETWEEN_REFRESHS, 5000); tableContext.typeChecker = typeChecker; - tableContext.defaultPrimeDocTerm = new Term("_primedoc_","true"); + tableContext.defaultPrimeDocTerm = new Term("_primedoc_", "true"); tableContext.defaultScoreType = ScoreType.SUPER; Class<?> c1 = configuration.getClass(BLUR_LUCENE_INDEX_DELETION_POLICY_CLASS, KeepOnlyLastCommitDeletionPolicy.class); @@ -101,6 +106,8 @@ public class TableContext { tableContext.similarity = (Similarity) configure(ReflectionUtils.newInstance(c2, configuration), tableContext); Class<?> c3 = configuration.getClass(BLUR_QUERY_CONVERTER_CLASS, QueryConverterImpl.class); tableContext.queryConverter = (QueryConverter) configure(ReflectionUtils.newInstance(c3, configuration), tableContext); + + cache.put(tableDescriptor.getName(), tableContext); return tableContext; } @@ -115,111 +122,59 @@ public class TableContext { return indexDeletionPolicy; } - public void setIndexDeletionPolicy(IndexDeletionPolicy indexDeletionPolicy) { - this.indexDeletionPolicy = indexDeletionPolicy; - } - public Similarity getSimilarity() { return similarity; } - public void setSimilarity(Similarity similarity) { - this.similarity = similarity; - } - public long getTimeBetweenCommits() { return timeBetweenCommits; } - public void setTimeBetweenCommits(long timeBetweenCommits) { - this.timeBetweenCommits = timeBetweenCommits; - } - public long getTimeBetweenRefreshs() { return timeBetweenRefreshs; } - public void setTimeBetweenRefreshs(long timeBetweenRefreshs) { - this.timeBetweenRefreshs = timeBetweenRefreshs; - } - public BlurAnalyzer getAnalyzer() { return analyzer; } - public void setAnalyzer(BlurAnalyzer analyzer) { - this.analyzer = analyzer; - } - public String getTable() { return table; } - public void setTable(String table) { - this.table = table; - } - public QueryConverter getQueryConverter() { return queryConverter; } - public void setQueryConverter(QueryConverter queryConverter) { - this.queryConverter = queryConverter; - } - public Configuration getConfiguration() { return configuration; } - public void setConfiguration(Configuration configuration) { - this.configuration = configuration; - } - public TableDescriptor getDescriptor() { return descriptor; } - public void setDescriptor(TableDescriptor descriptor) { - this.descriptor = descriptor; - } - public Path getTablePath() { return tablePath; } - public void setTablePath(Path tablePath) { - this.tablePath = tablePath; - } - public Path getWalTablePath() { return walTablePath; } - public void setWalTablePath(Path walTablePath) { - this.walTablePath = walTablePath; - } - public String getDefaultFieldName() { return defaultFieldName; } - public void setDefaultFieldName(String defaultFieldName) { - this.defaultFieldName = defaultFieldName; - } - public Term getDefaultPrimeDocTerm() { return defaultPrimeDocTerm; } - public void setDefaultScoreType(ScoreType defaultScoreType) { - this.defaultScoreType = defaultScoreType; - } - - public void setDefaultPrimeDocTerm(Term defaultPrimeDocTerm) { - this.defaultPrimeDocTerm = defaultPrimeDocTerm; - } - public ScoreType getDefaultScoreType() { return defaultScoreType; } + + public TypeManager getTypeManager() { + return typeChecker; + } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/cf92e0ff/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 86ce7d1..4db0232 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 @@ -243,6 +243,7 @@ public class ThriftBlurServer extends AbstractThriftServer { blurServer.setConfiguration(configuration); blurServer.setLayout(layout); blurServer.setNodeName(nodeName); + blurServer.setConf(config); blurServer.init(); Iface iface = BlurUtil.recordMethodCallsAndAverageTimes(blurMetrics, blurServer, Iface.class); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/cf92e0ff/src/blur-core/src/main/java/org/apache/blur/utils/BlurConstants.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/org/apache/blur/utils/BlurConstants.java b/src/blur-core/src/main/java/org/apache/blur/utils/BlurConstants.java index 17b107f..af376bb 100644 --- a/src/blur-core/src/main/java/org/apache/blur/utils/BlurConstants.java +++ b/src/blur-core/src/main/java/org/apache/blur/utils/BlurConstants.java @@ -46,6 +46,7 @@ public class BlurConstants { public static final String BLUR_CONTROLLER_HOSTNAME = "blur.controller.hostname"; public static final String BLUR_CONTROLLER_BIND_PORT = "blur.controller.bind.port"; public static final String BLUR_CONTROLLER_BIND_ADDRESS = "blur.controller.bind.address"; + public static final String BLUR_QUERYSTATUS_TRACKING_MAX = "blur.querystatus.tracking.max"; public static final String BLUR_QUERY_MAX_ROW_FETCH = "blur.query.max.row.fetch"; public static final String BLUR_QUERY_MAX_RECORD_FETCH = "blur.query.max.record.fetch"; public static final String BLUR_QUERY_MAX_RESULTS_FETCH = "blur.query.max.results.fetch"; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/cf92e0ff/src/blur-shell/src/main/java/org/apache/blur/shell/Main.java ---------------------------------------------------------------------- diff --git a/src/blur-shell/src/main/java/org/apache/blur/shell/Main.java b/src/blur-shell/src/main/java/org/apache/blur/shell/Main.java index c9cd397..47448b6 100644 --- a/src/blur-shell/src/main/java/org/apache/blur/shell/Main.java +++ b/src/blur-shell/src/main/java/org/apache/blur/shell/Main.java @@ -83,7 +83,7 @@ public class Main { } @Override - public void doit(PrintWriter out, String[] args, ConsoleReader reader) throws CommandException, TException, BlurException { + public void doit(PrintWriter out, String[] args, ConsoleReader reader) throws CommandException, TException, BlurException { if (timed == true) { timed = false; } else { @@ -185,7 +185,7 @@ public class Main { PrintWriter out = new PrintWriter(reader.getOutput()); try { while ((line = reader.readLine()) != null) { - + line = line.trim(); // ignore empty lines and comments if (line.length() == 0 || line.startsWith("#")) { @@ -198,7 +198,7 @@ public class Main { } else { long start = System.nanoTime(); try { - command.doit(out, commandArgs,reader); + command.doit(out, commandArgs, reader); } catch (QuitCommandException e) { // exit gracefully System.exit(0); @@ -240,12 +240,13 @@ public class Main { add(builder, new EnableTableCommand(client)); add(builder, new DisableTableCommand(client)); add(builder, new RemoveTableCommand(client)); + add(builder, new TruncateTableCommand(client)); add(builder, new DescribeCommand(client)); // builder.put("tablestats", new TableStatsCommand()); add(builder, new SchemaTableCommand(client)); add(builder, new SearchCommand(client)); add(builder, new AddCommand(client)); - // add(builder, new UpdateCommand(client)); + // add(builder, new UpdateCommand(client)); // builder.put("indexaccesslog", new IndexAccessLogCommand()); // builder.put("shardlayout", new ShardServerLayoutCommand()); commands = builder.build(); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/cf92e0ff/src/blur-shell/src/main/java/org/apache/blur/shell/SchemaTableCommand.java ---------------------------------------------------------------------- diff --git a/src/blur-shell/src/main/java/org/apache/blur/shell/SchemaTableCommand.java b/src/blur-shell/src/main/java/org/apache/blur/shell/SchemaTableCommand.java index cb17d95..031dbfe 100644 --- a/src/blur-shell/src/main/java/org/apache/blur/shell/SchemaTableCommand.java +++ b/src/blur-shell/src/main/java/org/apache/blur/shell/SchemaTableCommand.java @@ -27,7 +27,6 @@ import jline.console.ConsoleReader; import org.apache.blur.thrift.generated.Blur.Iface; import org.apache.blur.thrift.generated.BlurException; import org.apache.blur.thrift.generated.FieldSchema; -import org.apache.blur.thrift.generated.Session; import org.apache.blur.thrift.generated.TableSchema; import org.apache.thrift.TException; @@ -44,9 +43,7 @@ public class SchemaTableCommand extends TableSingleArgumentCommand { } String tablename = args[1]; - Session session = client.openReadSession(tablename); - TableSchema schema = client.schema(session); - client.closeReadSession(session); + TableSchema schema = client.schema(tablename); List<FieldSchema> fields = schema.getFields(); List<String> fieldNames = new ArrayList<String>();
