Repository: incubator-blur Updated Branches: refs/heads/master d92fbf629 -> 20c163205
Cleaning up the platform command api to allow for the combine method on the IndexReadCombiningCommand to be reused in the controller for less code creation. Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/20c16320 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/20c16320 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/20c16320 Branch: refs/heads/master Commit: 20c163205fe5c9a6288ac96356fd298546a7cda8 Parents: d92fbf6 Author: Aaron McCurry <[email protected]> Authored: Tue Sep 23 14:21:55 2014 -0400 Committer: Aaron McCurry <[email protected]> Committed: Tue Sep 23 14:21:55 2014 -0400 ---------------------------------------------------------------------- .../blur/command/DocumentCountCombiner.java | 2 +- .../DocumentCountDefaultClusterCombine.java | 50 ++++++++++++++++++++ .../org/apache/blur/command/TermsCommand.java | 3 +- .../blur/command/TestBlurObjectCommand.java | 7 +-- .../services/org.apache.blur.command.Commands | 3 +- .../apache/blur/command/BaseCommandManager.java | 3 +- .../command/ClusterReadCombiningCommand.java | 21 ++++++++ .../apache/blur/command/CombiningContext.java | 33 +++++++++++++ .../blur/command/ControllerCommandManager.java | 46 ++++++++++++++++-- .../blur/command/IndexReadCombiningCommand.java | 2 +- .../java/org/apache/blur/command/Location.java | 21 ++++++++ .../java/org/apache/blur/command/Server.java | 2 +- .../org/apache/blur/command/ServerContext.java | 33 ------------- .../java/org/apache/blur/command/Shard.java | 2 +- .../blur/command/ShardCommandManager.java | 6 +-- 15 files changed, 183 insertions(+), 51 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/20c16320/blur-command/src/main/java/org/apache/blur/command/DocumentCountCombiner.java ---------------------------------------------------------------------- diff --git a/blur-command/src/main/java/org/apache/blur/command/DocumentCountCombiner.java b/blur-command/src/main/java/org/apache/blur/command/DocumentCountCombiner.java index cf64ed8..4ade746 100644 --- a/blur-command/src/main/java/org/apache/blur/command/DocumentCountCombiner.java +++ b/blur-command/src/main/java/org/apache/blur/command/DocumentCountCombiner.java @@ -40,7 +40,7 @@ public class DocumentCountCombiner extends Command implements ClusterCommand<Lon } @Override - public Long combine(ServerContext context, Map<Shard, Integer> results) throws IOException { + public Long combine(CombiningContext context, Map<? extends Location<?>, Integer> results) throws IOException { long total = 0; for (Integer i : results.values()) { total += i; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/20c16320/blur-command/src/main/java/org/apache/blur/command/DocumentCountDefaultClusterCombine.java ---------------------------------------------------------------------- diff --git a/blur-command/src/main/java/org/apache/blur/command/DocumentCountDefaultClusterCombine.java b/blur-command/src/main/java/org/apache/blur/command/DocumentCountDefaultClusterCombine.java new file mode 100644 index 0000000..cc6d20c --- /dev/null +++ b/blur-command/src/main/java/org/apache/blur/command/DocumentCountDefaultClusterCombine.java @@ -0,0 +1,50 @@ +/** + * 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. + */ +package org.apache.blur.command; + +import java.io.IOException; +import java.util.Map; + +import org.apache.blur.command.annotation.Description; + +@SuppressWarnings("serial") +@Description("Gets the number of visible documents in the index.") +public class DocumentCountDefaultClusterCombine extends Command implements ClusterReadCombiningCommand<Integer, Long> { + + private static final String DOC_COUNT_CLUSTER_COMBINE = "docCountClusterCombine"; + + @Override + public String getName() { + return DOC_COUNT_CLUSTER_COMBINE; + } + + @Override + public Integer execute(IndexContext context) throws IOException { + return context.getIndexReader().numDocs(); + } + + @Override + public Long combine(CombiningContext context, Map<? extends Location<?>, Integer> results) throws IOException, + InterruptedException { + long total = 0; + for (Integer i : results.values()) { + total += i; + } + return total; + } + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/20c16320/blur-command/src/main/java/org/apache/blur/command/TermsCommand.java ---------------------------------------------------------------------- diff --git a/blur-command/src/main/java/org/apache/blur/command/TermsCommand.java b/blur-command/src/main/java/org/apache/blur/command/TermsCommand.java index 2530fbe..8e02fe6 100644 --- a/blur-command/src/main/java/org/apache/blur/command/TermsCommand.java +++ b/blur-command/src/main/java/org/apache/blur/command/TermsCommand.java @@ -56,7 +56,8 @@ public class TermsCommand extends Command implements ClusterCommand<List<String> } @Override - public List<String> combine(ServerContext context, Map<Shard, List<String>> results) throws IOException { + public List<String> combine(CombiningContext context, Map<? extends Location<?>, List<String>> results) + throws IOException, InterruptedException { TreeSet<String> terms = Sets.newTreeSet(); for (List<String> t : results.values()) { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/20c16320/blur-command/src/main/java/org/apache/blur/command/TestBlurObjectCommand.java ---------------------------------------------------------------------- diff --git a/blur-command/src/main/java/org/apache/blur/command/TestBlurObjectCommand.java b/blur-command/src/main/java/org/apache/blur/command/TestBlurObjectCommand.java index 63b11b8..421e030 100644 --- a/blur-command/src/main/java/org/apache/blur/command/TestBlurObjectCommand.java +++ b/blur-command/src/main/java/org/apache/blur/command/TestBlurObjectCommand.java @@ -32,11 +32,12 @@ public class TestBlurObjectCommand extends Command implements IndexReadCombining } @Override - public BlurObject combine(ServerContext context, Map<Shard, BlurObject> results) throws IOException { + public BlurObject combine(CombiningContext context, Map<? extends Location<?>, BlurObject> results) + throws IOException, InterruptedException { BlurObject blurObject = new BlurObject(); long total = 0; - for (Entry<Shard, BlurObject> e : results.entrySet()) { - total += e.getValue().getInteger("docCount"); + for (BlurObject bo : results.values()) { + total += bo.getInteger("docCount"); } blurObject.put("docCount", total); return blurObject; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/20c16320/blur-command/src/main/resources/META-INF/services/org.apache.blur.command.Commands ---------------------------------------------------------------------- diff --git a/blur-command/src/main/resources/META-INF/services/org.apache.blur.command.Commands b/blur-command/src/main/resources/META-INF/services/org.apache.blur.command.Commands index e10270e..7ffe0d2 100644 --- a/blur-command/src/main/resources/META-INF/services/org.apache.blur.command.Commands +++ b/blur-command/src/main/resources/META-INF/services/org.apache.blur.command.Commands @@ -13,4 +13,5 @@ # See the License for the specific language governing permissions and # limitations under the License. -org.apache.blur.command.DocumentCount \ No newline at end of file +org.apache.blur.command.DocumentCount +org.apache.blur.command.DocumentCountDefaultClusterCombine \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/20c16320/blur-core/src/main/java/org/apache/blur/command/BaseCommandManager.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/command/BaseCommandManager.java b/blur-core/src/main/java/org/apache/blur/command/BaseCommandManager.java index e246b8a..1042afd 100644 --- a/blur-core/src/main/java/org/apache/blur/command/BaseCommandManager.java +++ b/blur-core/src/main/java/org/apache/blur/command/BaseCommandManager.java @@ -470,7 +470,8 @@ public class BaseCommandManager implements Closeable { shardServerReturn = "shard->(" + returnType.getSimpleName() + ")"; } else if (command instanceof IndexReadCombiningCommand) { IndexReadCombiningCommand<?, ?> indexReadCombiningCommand = (IndexReadCombiningCommand<?, ?>) command; - Method method = indexReadCombiningCommand.getClass().getMethod("combine", new Class[] { Map.class }); + Method method = indexReadCombiningCommand.getClass().getMethod("combine", + new Class[] { CombiningContext.class, Map.class }); Class<?> returnType = method.getReturnType(); shardServerReturn = "server->(" + returnType.getSimpleName() + ")"; } else { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/20c16320/blur-core/src/main/java/org/apache/blur/command/ClusterReadCombiningCommand.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/command/ClusterReadCombiningCommand.java b/blur-core/src/main/java/org/apache/blur/command/ClusterReadCombiningCommand.java new file mode 100644 index 0000000..440da70 --- /dev/null +++ b/blur-core/src/main/java/org/apache/blur/command/ClusterReadCombiningCommand.java @@ -0,0 +1,21 @@ +/** + * 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. + */ +package org.apache.blur.command; + +public interface ClusterReadCombiningCommand<T1, T2> extends IndexReadCombiningCommand<T1, T2> { + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/20c16320/blur-core/src/main/java/org/apache/blur/command/CombiningContext.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/command/CombiningContext.java b/blur-core/src/main/java/org/apache/blur/command/CombiningContext.java new file mode 100644 index 0000000..b384b08 --- /dev/null +++ b/blur-core/src/main/java/org/apache/blur/command/CombiningContext.java @@ -0,0 +1,33 @@ +package org.apache.blur.command; + +import java.io.IOException; + +import org.apache.blur.BlurConfiguration; +import org.apache.blur.server.TableContext; + +/** + * 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. + */ + +public abstract class CombiningContext { + + public abstract Args getArgs(); + + public abstract TableContext getTableContext(String table) throws IOException; + + public abstract BlurConfiguration getBlurConfiguration(String table) throws IOException; + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/20c16320/blur-core/src/main/java/org/apache/blur/command/ControllerCommandManager.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/command/ControllerCommandManager.java b/blur-core/src/main/java/org/apache/blur/command/ControllerCommandManager.java index b30e23d..63dda47 100644 --- a/blur-core/src/main/java/org/apache/blur/command/ControllerCommandManager.java +++ b/blur-core/src/main/java/org/apache/blur/command/ControllerCommandManager.java @@ -4,7 +4,9 @@ import java.io.IOException; import java.util.Map; import java.util.concurrent.Callable; +import org.apache.blur.BlurConfiguration; import org.apache.blur.server.LayoutFactory; +import org.apache.blur.server.TableContext; import org.apache.blur.server.TableContextFactory; import org.apache.hadoop.conf.Configuration; @@ -32,8 +34,8 @@ public class ControllerCommandManager extends BaseCommandManager { super(tmpPath, commandPath, workerThreadCount, driverThreadCount, connectionTimeout, configuration); } - public Response execute(TableContextFactory tableContextFactory, LayoutFactory layoutFactory, String commandName, - final Args args) throws IOException, TimeoutException, ExceptionCollector { + public Response execute(final TableContextFactory tableContextFactory, LayoutFactory layoutFactory, + String commandName, final Args args) throws IOException, TimeoutException, ExceptionCollector { final ClusterContext context = createCommandContext(tableContextFactory, layoutFactory, args); final Command command = getCommandObject(commandName); if (command == null) { @@ -43,10 +45,12 @@ public class ControllerCommandManager extends BaseCommandManager { @Override public Response call() throws Exception { // For those commands that do not implement cluster command, run them in - // a - // base impl. + // a base impl. if (command instanceof ClusterCommand) { return executeClusterCommand(context, command); + } else if (command instanceof ClusterReadCombiningCommand) { + CombiningContext combiningContext = getCombiningContext(tableContextFactory, args); + return executeClusterReadCombiningCommand(args, context, command, combiningContext); } else if (command instanceof IndexReadCombiningCommand) { return executeIndexReadCombiningCommand(args, context, command); } else if (command instanceof IndexReadCommand) { @@ -55,10 +59,32 @@ public class ControllerCommandManager extends BaseCommandManager { throw new IOException("Command type of [" + command.getClass() + "] not supported."); } } + }); } - private Response executeClusterCommand(ClusterContext context, Command command) throws IOException, InterruptedException { + private CombiningContext getCombiningContext(final TableContextFactory tableContextFactory, final Args args) { + return new CombiningContext() { + + @Override + public TableContext getTableContext(String table) throws IOException { + return tableContextFactory.getTableContext(table); + } + + @Override + public BlurConfiguration getBlurConfiguration(String table) throws IOException { + return getTableContext(table).getBlurConfiguration(); + } + + @Override + public Args getArgs() { + return args; + } + }; + } + + private Response executeClusterCommand(ClusterContext context, Command command) throws IOException, + InterruptedException { ClusterCommand<Object> clusterCommand = (ClusterCommand<Object>) command; Object object = clusterCommand.clusterExecute(context); return Response.createNewAggregateResponse(object); @@ -70,6 +96,16 @@ public class ControllerCommandManager extends BaseCommandManager { return Response.createNewShardResponse(result); } + private Response executeClusterReadCombiningCommand(Args args, ClusterContext context, Command command, + CombiningContext combiningContext) throws IOException, InterruptedException { + Class<? extends ClusterReadCombiningCommand<Object, Object>> clazz = (Class<? extends ClusterReadCombiningCommand<Object, Object>>) command + .getClass(); + Map<Server, Object> results = context.readServers(args, clazz); + ClusterReadCombiningCommand<Object, Object> clusterReadCombiningCommand = (ClusterReadCombiningCommand<Object, Object>) command; + Object result = clusterReadCombiningCommand.combine(combiningContext, (Map<? extends Location<?>, Object>) results); + return Response.createNewAggregateResponse(result); + } + private Response executeIndexReadCombiningCommand(Args args, ClusterContext context, Command command) throws IOException { Class<? extends IndexReadCombiningCommand<Object, Object>> clazz = (Class<? extends IndexReadCombiningCommand<Object, Object>>) command http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/20c16320/blur-core/src/main/java/org/apache/blur/command/IndexReadCombiningCommand.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/command/IndexReadCombiningCommand.java b/blur-core/src/main/java/org/apache/blur/command/IndexReadCombiningCommand.java index f74c080..0e2314f 100644 --- a/blur-core/src/main/java/org/apache/blur/command/IndexReadCombiningCommand.java +++ b/blur-core/src/main/java/org/apache/blur/command/IndexReadCombiningCommand.java @@ -23,6 +23,6 @@ public interface IndexReadCombiningCommand<T1, T2> { T1 execute(IndexContext context) throws IOException, InterruptedException; - T2 combine(ServerContext context, Map<Shard, T1> results) throws IOException, InterruptedException; + T2 combine(CombiningContext context, Map<? extends Location<?>, T1> results) throws IOException, InterruptedException; } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/20c16320/blur-core/src/main/java/org/apache/blur/command/Location.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/command/Location.java b/blur-core/src/main/java/org/apache/blur/command/Location.java new file mode 100644 index 0000000..e54dc5e --- /dev/null +++ b/blur-core/src/main/java/org/apache/blur/command/Location.java @@ -0,0 +1,21 @@ +/** + * 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. + */ +package org.apache.blur.command; + +public abstract class Location<T> implements Comparable<T> { + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/20c16320/blur-core/src/main/java/org/apache/blur/command/Server.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/command/Server.java b/blur-core/src/main/java/org/apache/blur/command/Server.java index d559373..27a1300 100644 --- a/blur-core/src/main/java/org/apache/blur/command/Server.java +++ b/blur-core/src/main/java/org/apache/blur/command/Server.java @@ -17,7 +17,7 @@ package org.apache.blur.command; * the License. */ -public class Server implements Comparable<Server> { +public class Server extends Location<Server> implements Comparable<Server> { private final String _server; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/20c16320/blur-core/src/main/java/org/apache/blur/command/ServerContext.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/command/ServerContext.java b/blur-core/src/main/java/org/apache/blur/command/ServerContext.java deleted file mode 100644 index 81673b9..0000000 --- a/blur-core/src/main/java/org/apache/blur/command/ServerContext.java +++ /dev/null @@ -1,33 +0,0 @@ -package org.apache.blur.command; - -import java.io.IOException; - -import org.apache.blur.BlurConfiguration; -import org.apache.blur.server.TableContext; - -/** - * 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. - */ - -public abstract class ServerContext { - - public abstract Args getArgs(); - - public abstract TableContext getTableContext(String table) throws IOException; - - public abstract BlurConfiguration getBlurConfiguration(String table) throws IOException; - -} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/20c16320/blur-core/src/main/java/org/apache/blur/command/Shard.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/command/Shard.java b/blur-core/src/main/java/org/apache/blur/command/Shard.java index a431d92..a757512 100644 --- a/blur-core/src/main/java/org/apache/blur/command/Shard.java +++ b/blur-core/src/main/java/org/apache/blur/command/Shard.java @@ -17,7 +17,7 @@ package org.apache.blur.command; * the License. */ -public class Shard implements Comparable<Shard> { +public class Shard extends Location<Shard> implements Comparable<Shard> { private final String _shard; private final String _table; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/20c16320/blur-core/src/main/java/org/apache/blur/command/ShardCommandManager.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/command/ShardCommandManager.java b/blur-core/src/main/java/org/apache/blur/command/ShardCommandManager.java index b9565b5..8b64ca3 100644 --- a/blur-core/src/main/java/org/apache/blur/command/ShardCommandManager.java +++ b/blur-core/src/main/java/org/apache/blur/command/ShardCommandManager.java @@ -63,8 +63,8 @@ public class ShardCommandManager extends BaseCommandManager { throw new IOException("Command type of [" + command.getClass() + "] not supported."); } - private ServerContext getServerContext(final Args args, final TableContextFactory tableContextFactory) { - return new ServerContext() { + private CombiningContext getServerContext(final Args args, final TableContextFactory tableContextFactory) { + return new CombiningContext() { @Override public TableContext getTableContext(String table) throws IOException { @@ -95,7 +95,7 @@ public class ShardCommandManager extends BaseCommandManager { } @SuppressWarnings("unchecked") - private Response toResponse(Map<Shard, Object> results, Command command, ServerContext serverContext) + private Response toResponse(Map<Shard, Object> results, Command command, CombiningContext serverContext) throws IOException, InterruptedException { if (command instanceof IndexReadCombiningCommand) { IndexReadCombiningCommand<Object, Object> primitiveCommandAggregator = (IndexReadCombiningCommand<Object, Object>) command;
