Repository: incubator-blur Updated Branches: refs/heads/blur-platform 4d954b15e -> 27bcf1f93
Adding a new potential API for the blur platform command logic. This is just and API but there are some example usage of the API. Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/ba6564aa Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/ba6564aa Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/ba6564aa Branch: refs/heads/blur-platform Commit: ba6564aa6f64ddc75e90106a4413d267c0f305cf Parents: 067b8d4 Author: Aaron McCurry <[email protected]> Authored: Wed Aug 13 09:15:15 2014 -0400 Committer: Aaron McCurry <[email protected]> Committed: Wed Aug 13 09:15:15 2014 -0400 ---------------------------------------------------------------------- .../blur/server/platform/v2/Arguments.java | 32 +++ .../apache/blur/server/platform/v2/Client.java | 27 +++ .../blur/server/platform/v2/CommandContext.java | 25 ++ .../server/platform/v2/ControllerCommand.java | 35 +++ .../platform/v2/ControllerCommandContext.java | 42 ++++ .../blur/server/platform/v2/ReadCommand.java | 36 +++ .../server/platform/v2/ReadWriteCommand.java | 30 +++ .../apache/blur/server/platform/v2/Session.java | 30 +++ .../blur/server/platform/v2/ShardCommand.java | 36 +++ .../server/platform/v2/ShardCommandContext.java | 39 +++ .../v2/examples/GetTermsListStateless.java | 145 +++++++++++ .../v2/examples/GetTermsListStreaming.java | 239 +++++++++++++++++++ .../java/org/apache/blur/thrift/BException.java | 6 +- 13 files changed, 721 insertions(+), 1 deletion(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/ba6564aa/blur-core/src/main/java/org/apache/blur/server/platform/v2/Arguments.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/server/platform/v2/Arguments.java b/blur-core/src/main/java/org/apache/blur/server/platform/v2/Arguments.java new file mode 100644 index 0000000..9aa2c01 --- /dev/null +++ b/blur-core/src/main/java/org/apache/blur/server/platform/v2/Arguments.java @@ -0,0 +1,32 @@ +/** + * 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.server.platform.v2; + +import java.io.Serializable; + +@SuppressWarnings("serial") +public class Arguments implements Serializable, Cloneable { + + public <T> T get(String name) { + return null; + } + + public <T> Arguments set(String name, T value) { + return null; + } + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/ba6564aa/blur-core/src/main/java/org/apache/blur/server/platform/v2/Client.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/server/platform/v2/Client.java b/blur-core/src/main/java/org/apache/blur/server/platform/v2/Client.java new file mode 100644 index 0000000..e1ddeac --- /dev/null +++ b/blur-core/src/main/java/org/apache/blur/server/platform/v2/Client.java @@ -0,0 +1,27 @@ +/** + * 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.server.platform.v2; + +public class Client { + public <T> void setup(Arguments arguments, ControllerCommand<T> command) { + + } + + public <T> T execute(Arguments arguments, ControllerCommand<T> command) { + return null; + } +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/ba6564aa/blur-core/src/main/java/org/apache/blur/server/platform/v2/CommandContext.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/server/platform/v2/CommandContext.java b/blur-core/src/main/java/org/apache/blur/server/platform/v2/CommandContext.java new file mode 100644 index 0000000..2d615d5 --- /dev/null +++ b/blur-core/src/main/java/org/apache/blur/server/platform/v2/CommandContext.java @@ -0,0 +1,25 @@ +/** + * 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.server.platform.v2; + +import org.apache.blur.server.TableContext; + +public class CommandContext { + public TableContext getTableContext() { + return null; + } +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/ba6564aa/blur-core/src/main/java/org/apache/blur/server/platform/v2/ControllerCommand.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/server/platform/v2/ControllerCommand.java b/blur-core/src/main/java/org/apache/blur/server/platform/v2/ControllerCommand.java new file mode 100644 index 0000000..a172531 --- /dev/null +++ b/blur-core/src/main/java/org/apache/blur/server/platform/v2/ControllerCommand.java @@ -0,0 +1,35 @@ +/** + * 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.server.platform.v2; + +import java.io.IOException; +import java.io.Serializable; + +import org.apache.blur.thrift.generated.BlurException; + +@SuppressWarnings("serial") +public abstract class ControllerCommand<T> implements Serializable, Cloneable { + + public void setup(Arguments arguments, Session session, String table, ControllerCommandContext context) + throws BlurException, IOException { + + } + + public abstract T execute(Arguments arguments, Session session, String table, ControllerCommandContext context) + throws BlurException, IOException; + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/ba6564aa/blur-core/src/main/java/org/apache/blur/server/platform/v2/ControllerCommandContext.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/server/platform/v2/ControllerCommandContext.java b/blur-core/src/main/java/org/apache/blur/server/platform/v2/ControllerCommandContext.java new file mode 100644 index 0000000..15570c7 --- /dev/null +++ b/blur-core/src/main/java/org/apache/blur/server/platform/v2/ControllerCommandContext.java @@ -0,0 +1,42 @@ +/** + * 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.server.platform.v2; + +import java.util.Map; +import java.util.Set; +import java.util.concurrent.Future; + +public class ControllerCommandContext extends CommandContext { + + public Set<String> getAllShardIdsForTable(String table) { + return null; + } + + public <T> void setup(Arguments arguments, Session session, String table, ReadCommand<T> readCommand) { + + } + + public <T> Map<String, Future<T>> execute(Set<String> shardIdsToExecute, Arguments arguments, Session session, + ShardCommand<T> shardCommand) { + return null; + } + + public <T> Map<String, T> getValues(Map<String, Future<T>> results) { + return null; + } + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/ba6564aa/blur-core/src/main/java/org/apache/blur/server/platform/v2/ReadCommand.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/server/platform/v2/ReadCommand.java b/blur-core/src/main/java/org/apache/blur/server/platform/v2/ReadCommand.java new file mode 100644 index 0000000..423a782 --- /dev/null +++ b/blur-core/src/main/java/org/apache/blur/server/platform/v2/ReadCommand.java @@ -0,0 +1,36 @@ +/** + * 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.server.platform.v2; + +import java.io.IOException; +import java.io.Serializable; + +import org.apache.blur.server.IndexSearcherClosable; +import org.apache.blur.thrift.generated.BlurException; + +@SuppressWarnings("serial") +public abstract class ReadCommand<T> implements Serializable, Cloneable { + + public void setup(Arguments arguments, Session session, ShardCommandContext context, IndexSearcherClosable searcher) + throws BlurException, IOException { + + } + + public abstract T execute(Arguments arguments, Session session, ShardCommandContext context, + IndexSearcherClosable searcher) throws BlurException, IOException; + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/ba6564aa/blur-core/src/main/java/org/apache/blur/server/platform/v2/ReadWriteCommand.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/server/platform/v2/ReadWriteCommand.java b/blur-core/src/main/java/org/apache/blur/server/platform/v2/ReadWriteCommand.java new file mode 100644 index 0000000..6fbf9ca --- /dev/null +++ b/blur-core/src/main/java/org/apache/blur/server/platform/v2/ReadWriteCommand.java @@ -0,0 +1,30 @@ +/** + * 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.server.platform.v2; + +import java.io.Serializable; + +import org.apache.blur.server.IndexSearcherClosable; +import org.apache.blur.thrift.generated.BlurException; +import org.apache.lucene.index.IndexWriter; + +@SuppressWarnings("serial") +public abstract class ReadWriteCommand<T> implements Serializable { + + public abstract T execute(Session session, IndexWriter writer, IndexSearcherClosable searcher) throws BlurException; + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/ba6564aa/blur-core/src/main/java/org/apache/blur/server/platform/v2/Session.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/server/platform/v2/Session.java b/blur-core/src/main/java/org/apache/blur/server/platform/v2/Session.java new file mode 100644 index 0000000..af8816c --- /dev/null +++ b/blur-core/src/main/java/org/apache/blur/server/platform/v2/Session.java @@ -0,0 +1,30 @@ +/** + * 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.server.platform.v2; + + +public class Session { + + public <T> T get(String name) { + return null; + } + + public <T> T set(String name, T value) { + return value; + } + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/ba6564aa/blur-core/src/main/java/org/apache/blur/server/platform/v2/ShardCommand.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/server/platform/v2/ShardCommand.java b/blur-core/src/main/java/org/apache/blur/server/platform/v2/ShardCommand.java new file mode 100644 index 0000000..08bb96d --- /dev/null +++ b/blur-core/src/main/java/org/apache/blur/server/platform/v2/ShardCommand.java @@ -0,0 +1,36 @@ +/** + * 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.server.platform.v2; + +import java.io.IOException; +import java.io.Serializable; +import java.util.Set; + +import org.apache.blur.thrift.generated.BlurException; + +@SuppressWarnings("serial") +public abstract class ShardCommand<T> implements Serializable, Cloneable { + + public void setup(Arguments arguments, Session session, Set<String> shardIdsToExecute, ShardCommandContext context) + throws BlurException, IOException { + + } + + public abstract T execute(Arguments arguments, Session session, Set<String> shardIdsToExecute, + ShardCommandContext context) throws BlurException, IOException; + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/ba6564aa/blur-core/src/main/java/org/apache/blur/server/platform/v2/ShardCommandContext.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/server/platform/v2/ShardCommandContext.java b/blur-core/src/main/java/org/apache/blur/server/platform/v2/ShardCommandContext.java new file mode 100644 index 0000000..719921e --- /dev/null +++ b/blur-core/src/main/java/org/apache/blur/server/platform/v2/ShardCommandContext.java @@ -0,0 +1,39 @@ +/** + * 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.server.platform.v2; + +import java.util.Map; +import java.util.Set; +import java.util.concurrent.Future; + +public class ShardCommandContext extends CommandContext { + + // runs the setup in parallel + public <T> void setup(Arguments arguments, Session session, Set<String> shardIdsToExecute, ReadCommand<T> readCommand) { + + } + + public <T> Map<String, Future<T>> execute(Set<String> shardIdsToExecute, Arguments arguments, Session session, + ReadCommand<T> readCommand) { + return null; + } + + public <T> Map<String, T> getValues(Map<String, Future<T>> results) { + return null; + } + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/ba6564aa/blur-core/src/main/java/org/apache/blur/server/platform/v2/examples/GetTermsListStateless.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/server/platform/v2/examples/GetTermsListStateless.java b/blur-core/src/main/java/org/apache/blur/server/platform/v2/examples/GetTermsListStateless.java new file mode 100644 index 0000000..6d2a939 --- /dev/null +++ b/blur-core/src/main/java/org/apache/blur/server/platform/v2/examples/GetTermsListStateless.java @@ -0,0 +1,145 @@ +/** + * 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.server.platform.v2.examples; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.SortedSet; +import java.util.TreeSet; +import java.util.concurrent.Future; + +import org.apache.blur.server.IndexSearcherClosable; +import org.apache.blur.server.platform.v2.Arguments; +import org.apache.blur.server.platform.v2.Client; +import org.apache.blur.server.platform.v2.ControllerCommand; +import org.apache.blur.server.platform.v2.ControllerCommandContext; +import org.apache.blur.server.platform.v2.ReadCommand; +import org.apache.blur.server.platform.v2.Session; +import org.apache.blur.server.platform.v2.ShardCommand; +import org.apache.blur.server.platform.v2.ShardCommandContext; +import org.apache.blur.thrift.generated.BlurException; +import org.apache.lucene.index.AtomicReader; +import org.apache.lucene.index.IndexReader; +import org.apache.lucene.index.SlowCompositeReaderWrapper; +import org.apache.lucene.index.Terms; +import org.apache.lucene.index.TermsEnum; +import org.apache.lucene.util.BytesRef; + +public class GetTermsListStateless { + + private static final String FETCH = "FETCH"; + private static final String STARTS_WITH = "startsWith"; + private static final String FIELD = "field"; + + enum ACTION { + CURRENT, ADVANCE, NEXT + } + + @SuppressWarnings("serial") + public static void main(String[] args) { + + // Index logic + + final ReadCommand<List<BytesRef>> readCommand = new ReadCommand<List<BytesRef>>() { + @Override + public List<BytesRef> execute(Arguments arguments, Session session, ShardCommandContext context, IndexSearcherClosable searcher) + throws BlurException, IOException { + String field = arguments.get(FIELD); + BytesRef startsWith = arguments.get(STARTS_WITH); + int fetch = arguments.get(FETCH); + + IndexReader indexReader = searcher.getIndexReader(); + AtomicReader reader = SlowCompositeReaderWrapper.wrap(indexReader); + Terms terms = reader.fields().terms(field); + TermsEnum termsEnum = terms.iterator(null); + termsEnum.seekCeil(startsWith); + + List<BytesRef> results = new ArrayList<BytesRef>(); + results.add(termsEnum.term()); + BytesRef ref; + while ((ref = termsEnum.next()) != null) { + if (results.size() >= fetch) { + return results; + } + results.add(ref); + } + return results; + } + }; + + // Shard logic + + final ShardCommand<List<BytesRef>> shardCommand = new ShardCommand<List<BytesRef>>() { + + @Override + public List<BytesRef> execute(Arguments arguments, Session session, Set<String> shardIdsToExecute, + ShardCommandContext context) throws BlurException { + Map<String, Future<List<BytesRef>>> execute = context.execute(shardIdsToExecute, arguments, session, + readCommand); + Map<String, List<BytesRef>> values = context.getValues(execute); + SortedSet<BytesRef> resultSet = new TreeSet<BytesRef>(); + for (List<BytesRef> v : values.values()) { + resultSet.addAll(v); + } + List<BytesRef> results = new ArrayList<BytesRef>(); + results.addAll(resultSet); + int size = results.size(); + int fetch = arguments.get(FETCH); + return new ArrayList<BytesRef>(results.subList(0, Math.min(size, fetch))); + } + }; + + // Controller logic + + ControllerCommand<List<BytesRef>> controllerCommand = new ControllerCommand<List<BytesRef>>() { + @Override + public List<BytesRef> execute(Arguments arguments, Session session, String table, ControllerCommandContext context) + throws BlurException, IOException { + Set<String> shardIdsToExecute = context.getAllShardIdsForTable(table); + Map<String, Future<List<BytesRef>>> execute = context.execute(shardIdsToExecute, arguments, session, + shardCommand); + Map<String, List<BytesRef>> values = context.getValues(execute); + + SortedSet<BytesRef> resultSet = new TreeSet<BytesRef>(); + for (List<BytesRef> v : values.values()) { + resultSet.addAll(v); + } + List<BytesRef> results = new ArrayList<BytesRef>(); + results.addAll(resultSet); + int size = results.size(); + int fetch = arguments.get(FETCH); + return new ArrayList<BytesRef>(results.subList(0, Math.min(size, fetch))); + } + }; + + Client client = new Client(); + Arguments arguments = new Arguments().set(STARTS_WITH, new BytesRef("abc")); + List<BytesRef> refs; + while (!(refs = client.execute(arguments, controllerCommand)).isEmpty()) { + BytesRef last = null; + for (BytesRef ref : refs) { + last = ref; + System.out.println(ref.utf8ToString()); + } + arguments.set(STARTS_WITH, last); + } + + } +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/ba6564aa/blur-core/src/main/java/org/apache/blur/server/platform/v2/examples/GetTermsListStreaming.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/server/platform/v2/examples/GetTermsListStreaming.java b/blur-core/src/main/java/org/apache/blur/server/platform/v2/examples/GetTermsListStreaming.java new file mode 100644 index 0000000..0990eda --- /dev/null +++ b/blur-core/src/main/java/org/apache/blur/server/platform/v2/examples/GetTermsListStreaming.java @@ -0,0 +1,239 @@ +/** + * 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.server.platform.v2.examples; + +import java.io.IOException; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.Future; + +import org.apache.blur.analysis.FieldManager; +import org.apache.blur.analysis.FieldTypeDefinition; +import org.apache.blur.server.IndexSearcherClosable; +import org.apache.blur.server.TableContext; +import org.apache.blur.server.platform.v2.Arguments; +import org.apache.blur.server.platform.v2.Client; +import org.apache.blur.server.platform.v2.ControllerCommand; +import org.apache.blur.server.platform.v2.ControllerCommandContext; +import org.apache.blur.server.platform.v2.ReadCommand; +import org.apache.blur.server.platform.v2.Session; +import org.apache.blur.server.platform.v2.ShardCommand; +import org.apache.blur.server.platform.v2.ShardCommandContext; +import org.apache.blur.thrift.BException; +import org.apache.blur.thrift.generated.BlurException; +import org.apache.lucene.index.AtomicReader; +import org.apache.lucene.index.IndexReader; +import org.apache.lucene.index.SlowCompositeReaderWrapper; +import org.apache.lucene.index.Terms; +import org.apache.lucene.index.TermsEnum; +import org.apache.lucene.util.BytesRef; + +public class GetTermsListStreaming { + private static final String STARTS_WITH = "startsWith"; + private static final String FIELD_TYPE_DEFINITION = "fieldTypeDefinition"; + private static final String FIELD = "field"; + private static final String TERMS_ENUM = "termsEnum"; + private static final String CURRENT_TERM = "currentTerm"; + private static final String TERM = "term"; + private static final String ACTION_STR = "action"; + + enum ACTION { + CURRENT, ADVANCE, NEXT + } + + @SuppressWarnings("serial") + public static void main(String[] args) { + + // Index logic + + final ReadCommand<BytesRef> readCommand = new ReadCommand<BytesRef>() { + + @Override + public void setup(Arguments arguments, Session session, ShardCommandContext context, + IndexSearcherClosable searcher) throws BlurException, IOException { + TableContext tableContext = context.getTableContext(); + String field = arguments.get(FIELD); + + FieldManager fieldManager = tableContext.getFieldManager(); + FieldTypeDefinition fieldTypeDefinition = fieldManager.getFieldTypeDefinition(field); + + IndexReader indexReader = searcher.getIndexReader(); + AtomicReader reader = SlowCompositeReaderWrapper.wrap(indexReader); + Terms terms = reader.fields().terms(field); + TermsEnum termsEnum = terms.iterator(null); + + BytesRef ref = arguments.get(STARTS_WITH); + + termsEnum.seekCeil(ref); + + session.set(FIELD, field); + session.set(TERMS_ENUM, termsEnum); + session.set(FIELD_TYPE_DEFINITION, fieldTypeDefinition); + } + + @Override + public BytesRef execute(Arguments arguments, Session session, ShardCommandContext context, + IndexSearcherClosable searcher) throws BlurException, IOException { + TermsEnum termsEnum = session.get(TERMS_ENUM); + ACTION action = arguments.get(ACTION_STR); + BytesRef current = termsEnum.term(); + switch (action) { + case CURRENT: + return current; + case ADVANCE: { + BytesRef t = arguments.get(TERM); + if (t.compareTo(current) <= 0) { + return current; + } + termsEnum.seekCeil(t); + return termsEnum.term(); + } + case NEXT: { + throw new RuntimeException("Not impl yet."); + } + default: + throw new BException("Unknown action [{0}]", action); + } + } + }; + + // Shard logic + + final ShardCommand<BytesRef> shardCommand = new ShardCommand<BytesRef>() { + + @Override + public void setup(Arguments arguments, Session session, Set<String> shardIdsToExecute, ShardCommandContext context) + throws BlurException, IOException { + context.setup(arguments, session, shardIdsToExecute, readCommand); + Arguments argument = new Arguments().set(ACTION_STR, ACTION.CURRENT); + BytesRef lowest = executeAndGetLowestResult(readCommand, session, shardIdsToExecute, context, argument); + session.set(CURRENT_TERM, lowest); + } + + private BytesRef executeAndGetLowestResult(final ReadCommand<BytesRef> readCommand, Session session, + Set<String> shardIdsToExecute, ShardCommandContext context, Arguments argument) { + Map<String, Future<BytesRef>> results = context.execute(shardIdsToExecute, argument, session, readCommand); + Map<String, BytesRef> values = context.getValues(results); + // get lowest term + BytesRef lowest = null; + for (BytesRef bytesRef : values.values()) { + if (bytesRef == null) { + continue; + } else if (lowest == null || bytesRef.compareTo(lowest) < 0) { + lowest = bytesRef; + } + } + return lowest; + } + + @Override + public BytesRef execute(Arguments arguments, Session session, Set<String> shardIdsToExecute, + ShardCommandContext context) throws BlurException { + BytesRef current = session.get(CURRENT_TERM); + + BytesRef term = arguments.get(TERM); + ACTION action = arguments.get(ACTION_STR); + + switch (action) { + case CURRENT: { + return current; + } + case ADVANCE: { + Arguments argument = new Arguments().set(ACTION_STR, ACTION.ADVANCE).set(TERM, term); + BytesRef lowest = executeAndGetLowestResult(readCommand, session, shardIdsToExecute, context, argument); + return session.set(CURRENT_TERM, lowest); + } + case NEXT: { + throw new RuntimeException("Not impl yet."); + } + default: + throw new BException("Unknown action [{0}]", action); + } + } + }; + + // Controller logic + + ControllerCommand<BytesRef> controllerCommand = new ControllerCommand<BytesRef>() { + + @Override + public void setup(Arguments arguments, Session session, String table, ControllerCommandContext context) + throws BlurException, IOException { + context.setup(arguments, session, table, readCommand); + + Set<String> shardIdsToExecute = context.getAllShardIdsForTable(table); + Arguments argument = new Arguments().set(ACTION_STR, ACTION.CURRENT); + BytesRef lowest = executeAndGetLowestResult(readCommand, session, shardIdsToExecute, context, argument); + session.set(CURRENT_TERM, lowest); + } + + @Override + public BytesRef execute(Arguments arguments, Session session, String table, ControllerCommandContext context) + throws BlurException, IOException { + Set<String> shardIdsToExecute = context.getAllShardIdsForTable(table); + + BytesRef current = session.get(CURRENT_TERM); + + BytesRef term = arguments.get(TERM); + ACTION action = arguments.get(ACTION_STR); + + switch (action) { + case CURRENT: { + return current; + } + case ADVANCE: { + Arguments argument = new Arguments().set(ACTION_STR, ACTION.ADVANCE).set(TERM, term); + BytesRef lowest = executeAndGetLowestResult(readCommand, session, shardIdsToExecute, context, argument); + return session.set(CURRENT_TERM, lowest); + } + case NEXT: { + throw new RuntimeException("Not impl yet."); + } + default: + throw new BException("Unknown action [{0}]", action); + } + + } + + private BytesRef executeAndGetLowestResult(final ReadCommand<BytesRef> readCommand, Session session, + Set<String> shardIdsToExecute, ControllerCommandContext context, Arguments argument) { + Map<String, Future<BytesRef>> results = context.execute(shardIdsToExecute, argument, session, shardCommand); + Map<String, BytesRef> values = context.getValues(results); + // get lowest term + BytesRef lowest = null; + for (BytesRef bytesRef : values.values()) { + if (bytesRef == null) { + continue; + } else if (lowest == null || bytesRef.compareTo(lowest) < 0) { + lowest = bytesRef; + } + } + return lowest; + } + }; + + Client client = new Client(); + client.setup(new Arguments().set(FIELD, "f1").set(STARTS_WITH, new BytesRef("abc")), controllerCommand); + + Arguments arguments = new Arguments().set(ACTION_STR, ACTION.NEXT); + BytesRef ref; + while ((ref = client.execute(arguments, controllerCommand)) != null) { + System.out.println(ref.utf8ToString()); + } + + } +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/ba6564aa/blur-thrift/src/main/java/org/apache/blur/thrift/BException.java ---------------------------------------------------------------------- diff --git a/blur-thrift/src/main/java/org/apache/blur/thrift/BException.java b/blur-thrift/src/main/java/org/apache/blur/thrift/BException.java index 40fad0c..6f4acf8 100644 --- a/blur-thrift/src/main/java/org/apache/blur/thrift/BException.java +++ b/blur-thrift/src/main/java/org/apache/blur/thrift/BException.java @@ -17,13 +17,13 @@ package org.apache.blur.thrift; * limitations under the License. */ import java.io.ByteArrayOutputStream; +import java.io.IOException; import java.io.PrintWriter; import java.text.MessageFormat; import org.apache.blur.thrift.generated.BlurException; import org.apache.blur.thrift.generated.ErrorType; - public class BException extends BlurException { private static final long serialVersionUID = 5846541677293727358L; @@ -50,4 +50,8 @@ public class BException extends BlurException { writer.close(); return new String(outputStream.toByteArray()); } + + public static BlurException wrap(IOException e) { + return new BException(e.getMessage(), e); + } }
