BLUR-ID:50 Added validation for non indexable fields. Signed-off-by: Aaron McCurry <[email protected]>
Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/35f71a05 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/35f71a05 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/35f71a05 Branch: refs/heads/0.2-dev Commit: 35f71a05ac9465ea68618569ce6696cd6635e6e2 Parents: fdb5f48 Author: Gagan <[email protected]> Authored: Sat Jan 5 12:03:28 2013 +0530 Committer: Aaron McCurry <[email protected]> Committed: Sun Jan 6 18:21:32 2013 -0500 ---------------------------------------------------------------------- .../java/org/apache/blur/thrift/BlurServer.java | 16 +++-- .../org/apache/blur/utils/BlurValidations.java | 62 +++++++++++++++ .../org/apache/blur/mapreduce/lib/BlurWriter.java | 32 +++++--- 3 files changed, 93 insertions(+), 17 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/35f71a05/src/blur-core/src/main/java/org/apache/blur/thrift/BlurServer.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/org/apache/blur/thrift/BlurServer.java b/src/blur-core/src/main/java/org/apache/blur/thrift/BlurServer.java index c148d3b..4a0b21e 100644 --- a/src/blur-core/src/main/java/org/apache/blur/thrift/BlurServer.java +++ b/src/blur-core/src/main/java/org/apache/blur/thrift/BlurServer.java @@ -53,6 +53,7 @@ import org.apache.blur.thrift.generated.Blur.Client; import org.apache.blur.thrift.generated.Blur.Iface; import org.apache.blur.thrift.generated.BlurException; import org.apache.blur.thrift.generated.Document; +import org.apache.blur.thrift.generated.Field; import org.apache.blur.thrift.generated.Generation; import org.apache.blur.thrift.generated.LiveSchema; import org.apache.blur.thrift.generated.MutateOptions; @@ -68,6 +69,7 @@ import org.apache.blur.thrift.generated.TopFieldDocs; import org.apache.blur.thrift.generated.UpdatePackage; import org.apache.blur.utils.BlurConstants; import org.apache.blur.utils.BlurUtil; +import org.apache.blur.utils.BlurValidations; import org.apache.blur.utils.ThriftLuceneConversion; import org.apache.lucene.index.IndexReader; import org.apache.lucene.index.IndexableField; @@ -368,10 +370,11 @@ public class BlurServer extends TableAdmin implements Iface { List<Generation> generations = new ArrayList<Generation>(); try { BlurIndex index = getIndex(table, shardIndex); + List<Document> indexableDocuments = BlurValidations.getAllIndexableDocuments(documents); if (index == null) { - generations.addAll(forwardAddDocuments(options, documents)); + generations.addAll(forwardAddDocuments(options, indexableDocuments)); } else { - long generation = index.addDocuments(waitToBeVisible, writeAheadLog, documents); + long generation = index.addDocuments(waitToBeVisible, writeAheadLog, indexableDocuments); generations.add(new Generation(table, shardIndex, generation)); } return generations; @@ -380,7 +383,7 @@ public class BlurServer extends TableAdmin implements Iface { throw new BException(t.getMessage(), t); } } - + private int getShardIndex(MutateOptions options) { int shardIndex = options.getShardIndex(); if (shardIndex < 0) { @@ -446,7 +449,7 @@ public class BlurServer extends TableAdmin implements Iface { throw new BException(t.getMessage(), t); } } - + @Override public List<Generation> deleteDocuments(MutateOptions options, List<Term> terms) throws BlurException, TException { String table = options.getTable(); @@ -478,10 +481,11 @@ public class BlurServer extends TableAdmin implements Iface { List<Generation> generations = new ArrayList<Generation>(); try { BlurIndex index = getIndex(table, shardIndex); + List<UpdatePackage> updatePackagesWithIndexableDocs = BlurValidations.getAllIndexablePackages(updatePackages); if (index == null) { - generations.addAll(forwardUpdateDocuments(options, updatePackages)); + generations.addAll(forwardUpdateDocuments(options, updatePackagesWithIndexableDocs)); } else { - long generation = index.updateDocuments(waitToBeVisible, writeAheadLog, updatePackages); + long generation = index.updateDocuments(waitToBeVisible, writeAheadLog, updatePackagesWithIndexableDocs); generations.add(new Generation(table, shardIndex, generation)); } return generations; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/35f71a05/src/blur-core/src/main/java/org/apache/blur/utils/BlurValidations.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/org/apache/blur/utils/BlurValidations.java b/src/blur-core/src/main/java/org/apache/blur/utils/BlurValidations.java new file mode 100644 index 0000000..5ae5140 --- /dev/null +++ b/src/blur-core/src/main/java/org/apache/blur/utils/BlurValidations.java @@ -0,0 +1,62 @@ +package org.apache.blur.utils; + +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import java.util.Iterator; +import java.util.List; + +import org.apache.blur.thrift.generated.Document; +import org.apache.blur.thrift.generated.Field; +import org.apache.blur.thrift.generated.TYPE; +import org.apache.blur.thrift.generated.UpdatePackage; + +public class BlurValidations { + + public static List<UpdatePackage> getAllIndexablePackages(List<UpdatePackage> updatePackages) { + for(UpdatePackage updatePackage : updatePackages){ + List<Document> documents = getAllIndexableDocuments(updatePackage.getDocuments()); + updatePackage.setDocuments(documents); + } + return updatePackages; + } + + public static List<Document> getAllIndexableDocuments(List<Document> documents) { + Iterator<Document> itr = documents.iterator(); + while(itr.hasNext()){ + boolean isDocumentIndexable = checkIfAnyFieldIsIndexable(itr.next()); + if(!isDocumentIndexable){ + itr.remove(); + } + } + return documents; + } + + public static boolean checkIfAnyFieldIsIndexable(Document document) { + Iterator<Field> fields = document.getFieldsIterator(); + while (fields.hasNext()) { + Field field = fields.next(); + if (field.isSetType()) { + if (field.getType() != TYPE.BINARY) { + return true; + } + } + } + return false; + } + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/35f71a05/src/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurWriter.java ---------------------------------------------------------------------- diff --git a/src/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurWriter.java b/src/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurWriter.java index a5be0be..8043c71 100644 --- a/src/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurWriter.java +++ b/src/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurWriter.java @@ -19,6 +19,7 @@ import org.apache.blur.thrift.generated.Document; import org.apache.blur.thrift.generated.MutateOptions; import org.apache.blur.thrift.generated.Term; import org.apache.blur.thrift.generated.UpdatePackage; +import org.apache.blur.utils.BlurValidations; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.io.Writable; import org.apache.hadoop.mapreduce.Counter; @@ -37,14 +38,18 @@ public class BlurWriter extends RecordWriter<Writable, MutateWritable> { private Counter updateCounter; private Counter addCounter; - public BlurWriter(Configuration configuration, int shardIndex, Counter rateCounter, Counter documentCounter, Counter fieldCounter, Counter addCounter, Counter updateCounter, - Counter deleteCounter) { + public BlurWriter(Configuration configuration, int shardIndex, + Counter rateCounter, Counter documentCounter, Counter fieldCounter, + Counter addCounter, Counter updateCounter, Counter deleteCounter) { String connectionStr = configuration.get(BLUR_CONNECTION_STR); String table = configuration.get(BLUR_TABLE_NAME); - boolean writeAheadLog = configuration.getBoolean(BLUR_WRITE_AHEAD_LOG, false); - boolean waitToBeVisible = configuration.getBoolean(BLUR_WAIT_TO_BE_VISIBLE, false); + boolean writeAheadLog = configuration.getBoolean(BLUR_WRITE_AHEAD_LOG, + false); + boolean waitToBeVisible = configuration.getBoolean(BLUR_WAIT_TO_BE_VISIBLE, + false); client = BlurClient.getClient(connectionStr); - options = new MutateOptions(table, shardIndex, waitToBeVisible, writeAheadLog); + options = new MutateOptions(table, shardIndex, waitToBeVisible, + writeAheadLog); this.rateCounter = rateCounter; this.documentCounter = documentCounter; this.fieldCounter = fieldCounter; @@ -55,14 +60,16 @@ public class BlurWriter extends RecordWriter<Writable, MutateWritable> { @SuppressWarnings("unchecked") @Override - public void write(Writable key, MutateWritable value) throws IOException, InterruptedException { + public void write(Writable key, MutateWritable value) throws IOException, + InterruptedException { if (value == null) { return; } if (value instanceof AddDocumentsWritable) { AddDocumentsWritable addDocumentsWritable = (AddDocumentsWritable) value; try { - client.addDocuments(options, countAdds((List<Document>) addDocumentsWritable.getDocuments())); + List<Document> documents = BlurValidations.getAllIndexableDocuments((List<Document>)addDocumentsWritable.getDocuments()); + client.addDocuments(options,countAdds(documents)); } catch (BlurException e) { throw new IOException(e); } catch (TException e) { @@ -71,7 +78,7 @@ public class BlurWriter extends RecordWriter<Writable, MutateWritable> { } else if (value instanceof DeleteTermsWritable) { DeleteTermsWritable deleteTermsWritable = (DeleteTermsWritable) value; try { - client.deleteDocuments(options, countDeletes((List<Term>) deleteTermsWritable.getTerms())); + client.deleteDocuments(options,countDeletes((List<Term>) deleteTermsWritable.getTerms())); } catch (BlurException e) { throw new IOException(e); } catch (TException e) { @@ -79,15 +86,17 @@ public class BlurWriter extends RecordWriter<Writable, MutateWritable> { } } else if (value instanceof UpdatePackagesWritable) { UpdatePackagesWritable updatePackagesWritable = (UpdatePackagesWritable) value; + List<UpdatePackage> updatePackages = BlurValidations.getAllIndexablePackages((List<UpdatePackage>) updatePackagesWritable.getUpdatePackages()); try { - client.updateDocuments(options, countUpdates((List<UpdatePackage>) updatePackagesWritable.getUpdatePackages())); + client.updateDocuments(options, countUpdates(updatePackages)); } catch (BlurException e) { throw new IOException(e); } catch (TException e) { throw new IOException(e); } } else { - throw new RuntimeException("MutateWritable type of [" + value.getClass() + "] in instance [" + value + "] not supported."); + throw new RuntimeException("MutateWritable type of [" + value.getClass() + + "] in instance [" + value + "] not supported."); } } @@ -119,7 +128,8 @@ public class BlurWriter extends RecordWriter<Writable, MutateWritable> { } @Override - public void close(TaskAttemptContext context) throws IOException, InterruptedException { + public void close(TaskAttemptContext context) throws IOException, + InterruptedException { rateCounter.setValue(0); context.progress(); }
