| /* |
| * 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.rya.indexing.accumulo.freetext; |
| |
| import static java.util.Objects.requireNonNull; |
| import static org.apache.rya.indexing.accumulo.freetext.query.ASTNodeUtils.getNodeIterator; |
| |
| import java.io.IOException; |
| import java.nio.charset.CharacterCodingException; |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map.Entry; |
| import java.util.Objects; |
| import java.util.Set; |
| import java.util.SortedSet; |
| import java.util.TreeSet; |
| |
| import org.apache.accumulo.core.client.AccumuloException; |
| import org.apache.accumulo.core.client.AccumuloSecurityException; |
| import org.apache.accumulo.core.client.BatchWriter; |
| import org.apache.accumulo.core.client.Connector; |
| import org.apache.accumulo.core.client.IteratorSetting; |
| import org.apache.accumulo.core.client.MultiTableBatchWriter; |
| import org.apache.accumulo.core.client.MutationsRejectedException; |
| import org.apache.accumulo.core.client.Scanner; |
| import org.apache.accumulo.core.client.TableExistsException; |
| import org.apache.accumulo.core.client.TableNotFoundException; |
| import org.apache.accumulo.core.client.admin.TableOperations; |
| import org.apache.accumulo.core.data.Key; |
| import org.apache.accumulo.core.data.Mutation; |
| import org.apache.accumulo.core.data.Range; |
| import org.apache.accumulo.core.data.Value; |
| import org.apache.accumulo.core.file.keyfunctor.ColumnFamilyFunctor; |
| import org.apache.accumulo.core.iterators.user.IntersectingIterator; |
| import org.apache.commons.lang.StringUtils; |
| import org.apache.commons.lang.Validate; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.io.Text; |
| import org.apache.log4j.Logger; |
| import org.apache.rya.accumulo.experimental.AbstractAccumuloIndexer; |
| import org.apache.rya.api.RdfCloudTripleStoreConfiguration; |
| import org.apache.rya.api.domain.RyaStatement; |
| import org.apache.rya.api.resolver.RyaToRdfConversions; |
| import org.apache.rya.indexing.FreeTextIndexer; |
| import org.apache.rya.indexing.Md5Hash; |
| import org.apache.rya.indexing.StatementConstraints; |
| import org.apache.rya.indexing.StatementSerializer; |
| import org.apache.rya.indexing.accumulo.ConfigUtils; |
| import org.apache.rya.indexing.accumulo.freetext.iterators.BooleanTreeIterator; |
| import org.apache.rya.indexing.accumulo.freetext.query.ASTExpression; |
| import org.apache.rya.indexing.accumulo.freetext.query.ASTNodeUtils; |
| import org.apache.rya.indexing.accumulo.freetext.query.ASTSimpleNode; |
| import org.apache.rya.indexing.accumulo.freetext.query.ASTTerm; |
| import org.apache.rya.indexing.accumulo.freetext.query.ParseException; |
| import org.apache.rya.indexing.accumulo.freetext.query.QueryParser; |
| import org.apache.rya.indexing.accumulo.freetext.query.QueryParserTreeConstants; |
| import org.apache.rya.indexing.accumulo.freetext.query.SimpleNode; |
| import org.apache.rya.indexing.accumulo.freetext.query.TokenMgrError; |
| import org.eclipse.rdf4j.common.iteration.CloseableIteration; |
| import org.eclipse.rdf4j.model.IRI; |
| import org.eclipse.rdf4j.model.Literal; |
| import org.eclipse.rdf4j.model.Statement; |
| import org.eclipse.rdf4j.query.QueryEvaluationException; |
| |
| import com.google.common.base.Charsets; |
| import com.google.common.collect.Lists; |
| |
| /** |
| * The {@link AccumuloFreeTextIndexer} stores and queries "free text" data from statements into tables in Accumulo. Specifically, this class |
| * stores data into two different Accumulo Tables. This is the <b>document table</b> (default name: triplestore_text) and the <b>terms |
| * table</b> (default name: triplestore_terms). |
| * <p> |
| * The document table stores the document (i.e. a triple statement), document properties, and the terms within the document. This is the |
| * main table used for processing a text search by using document partitioned indexing. See {@link IntersectingIterator}. |
| * <p> |
| * For each document, the document table will store the following information: |
| * <P> |
| * |
| * <pre> |
| * Row (partition) | Column Family | Column Qualifier | Value |
| * ================+================+==================+========== |
| * shardID | d\x00 | documentHash | Document |
| * shardID | s\x00Subject | documentHash | (empty) |
| * shardID | p\x00Predicate | documentHash | (empty) |
| * shardID | o\x00Object | documentHash | (empty) |
| * shardID | c\x00Context | documentHash | (empty) |
| * shardID | t\x00token | documentHash | (empty) |
| * </pre> |
| * <p> |
| * Note: documentHash is a sha256 Hash of the Document's Content |
| * <p> |
| * The terms table is used for expanding wildcard search terms. For each token in the document table, the table will store the following |
| * information: |
| * |
| * <pre> |
| * Row (partition) | CF/CQ/Value |
| * ==================+============= |
| * l\x00token | (empty) |
| * r\x00Reversetoken | (empty) |
| * </pre> |
| * <p> |
| * There are two prefixes in the table, "token list" (keys with an "l" prefix) and "reverse token list" (keys with a "r" prefix). This table |
| * is uses the "token list" to expand foo* into terms like food, foot, and football. This table uses the "reverse token list" to expand *ar |
| * into car, bar, and far. |
| * <p> |
| * Example: Given these three statements as inputs: |
| * |
| * <pre> |
| * <uri:paul> rdfs:label "paul smith"@en <uri:graph1> |
| * <uri:steve> rdfs:label "steven anthony miller"@en <uri:graph1> |
| * <uri:steve> rdfs:label "steve miller"@en <uri:graph1> |
| * </pre> |
| * <p> |
| * Here's what the tables would look like: (Note: the hashes aren't real, the rows are not sorted, and the partition ids will vary.) |
| * <p> |
| * Triplestore_text |
| * |
| * <pre> |
| * Row (partition) | Column Family | Column Qualifier | Value |
| * ================+=================================+==================+========== |
| * 000000 | d\x00 | 08b3d233a | uri:graph1x00uri:paul\x00rdfs:label\x00"paul smith"@en |
| * 000000 | s\x00uri:paul | 08b3d233a | (empty) |
| * 000000 | p\x00rdfs:label | 08b3d233a | (empty) |
| * 000000 | o\x00"paul smith"@en | 08b3d233a | (empty) |
| * 000000 | c\x00uri:graph1 | 08b3d233a | (empty) |
| * 000000 | t\x00paul | 08b3d233a | (empty) |
| * 000000 | t\x00smith | 08b3d233a | (empty) |
| * |
| * 000000 | d\x00 | 3a575534b | uri:graph1x00uri:steve\x00rdfs:label\x00"steven anthony miller"@en |
| * 000000 | s\x00uri:steve | 3a575534b | (empty) |
| * 000000 | p\x00rdfs:label | 3a575534b | (empty) |
| * 000000 | o\x00"steven anthony miller"@en | 3a575534b | (empty) |
| * 000000 | c\x00uri:graph1 | 3a575534b | (empty) |
| * 000000 | t\x00steven | 3a575534b | (empty) |
| * 000000 | t\x00anthony | 3a575534b | (empty) |
| * 000000 | t\x00miller | 3a575534b | (empty) |
| * |
| * 000001 | d\x00 | 7bf670d06 | uri:graph1x00uri:steve\x00rdfs:label\x00"steve miller"@en |
| * 000001 | s\x00uri:steve | 7bf670d06 | (empty) |
| * 000001 | p\x00rdfs:label | 7bf670d06 | (empty) |
| * 000001 | o\x00"steve miller"@en | 7bf670d06 | (empty) |
| * 000001 | c\x00uri:graph1 | 7bf670d06 | (empty) |
| * 000001 | t\x00steve | 7bf670d06 | (empty) |
| * 000001 | t\x00miller | 7bf670d06 | (empty) |
| * </pre> |
| * <p> |
| * triplestore_terms |
| * <p> |
| * |
| * <pre> |
| * Row (partition) | CF/CQ/Value |
| * ==================+============= |
| * l\x00paul | (empty) |
| * l\x00smith | (empty) |
| * l\x00steven | (empty) |
| * l\x00anthony | (empty) |
| * l\x00miller | (empty) |
| * l\x00steve | (empty) |
| * r\x00luap | (empty) |
| * r\x00htims | (empty) |
| * r\x00nevets | (empty) |
| * r\x00ynohtna | (empty) |
| * r\x00rellim | (empty) |
| * r\x00evets | (empty) |
| * |
| * <pre> |
| */ |
| public class AccumuloFreeTextIndexer extends AbstractAccumuloIndexer implements FreeTextIndexer { |
| private static final String TABLE_SUFFIX_TERM = "freetext_term"; |
| |
| private static final String TABLE_SUFFFIX_DOC = "freetext"; |
| |
| private static final Logger logger = Logger.getLogger(AccumuloFreeTextIndexer.class); |
| |
| private static final boolean IS_TERM_TABLE_TOKEN_DELETION_ENABLED = true; |
| |
| private static final byte[] EMPTY_BYTES = new byte[] {}; |
| private static final Text EMPTY_TEXT = new Text(EMPTY_BYTES); |
| private static final Value EMPTY_VALUE = new Value(EMPTY_BYTES); |
| |
| private Tokenizer tokenizer; |
| |
| private BatchWriter docTableBw; |
| private BatchWriter termTableBw; |
| private MultiTableBatchWriter mtbw; |
| |
| private int queryTermLimit; |
| |
| private int docTableNumPartitions; |
| |
| private Set<IRI> validPredicates; |
| |
| private Configuration conf; |
| |
| private boolean isInit = false; |
| |
| /** |
| * Called by setConf to initialize query only. |
| * Use this alone if usage does not require writing. |
| * For a writeable (store and delete) version of this, |
| * call setconf() and then setMultiTableBatchWriter(), then call init() |
| * that is what the DAO does. |
| * @throws AccumuloException |
| * @throws AccumuloSecurityException |
| * @throws TableNotFoundException |
| * @throws TableExistsException |
| */ |
| private void initInternal() throws AccumuloException, AccumuloSecurityException, TableNotFoundException, |
| TableExistsException { |
| final String doctable = getFreeTextDocTablename(conf); |
| final String termtable = getFreeTextTermTablename(conf); |
| |
| docTableNumPartitions = ConfigUtils.getFreeTextDocNumPartitions(conf); |
| final int termTableNumPartitions = ConfigUtils.getFreeTextTermNumPartitions(conf); |
| |
| final TableOperations tableOps = ConfigUtils.getConnector(conf).tableOperations(); |
| |
| // Create term table partitions |
| final boolean createdTermTable = ConfigUtils.createTableIfNotExists(conf, termtable); |
| if (createdTermTable && !ConfigUtils.useMockInstance(conf) && termTableNumPartitions > 0) { |
| final TreeSet<Text> splits = new TreeSet<Text>(); |
| |
| // split on the "Term List" and "Reverse Term list" boundary |
| splits.add(new Text(ColumnPrefixes.getRevTermListColFam(""))); |
| |
| // Symmetrically split the "Term List" and "Reverse Term list" |
| final int numSubpartitions = ((termTableNumPartitions - 1) / 2); |
| if (numSubpartitions > 0) { |
| final int step = (26 / numSubpartitions); |
| for (int i = 0; i < numSubpartitions; i++) { |
| final String nextChar = String.valueOf((char) ('a' + (step * i))); |
| splits.add(new Text(ColumnPrefixes.getTermListColFam(nextChar))); |
| splits.add(new Text(ColumnPrefixes.getRevTermListColFam(nextChar))); |
| } |
| } |
| tableOps.addSplits(termtable, splits); |
| } |
| |
| // Create document (text) table partitions |
| final boolean createdDocTable = ConfigUtils.createTableIfNotExists(conf, doctable); |
| if (createdDocTable && !ConfigUtils.useMockInstance(conf)) { |
| final TreeSet<Text> splits = new TreeSet<Text>(); |
| for (int i = 0; i < docTableNumPartitions; i++) { |
| splits.add(genPartition(i, docTableNumPartitions)); |
| } |
| tableOps.addSplits(doctable, splits); |
| |
| // Add a tablet level Bloom filter for the Column Family. |
| // This will allow us to quickly determine if a term is contained in a tablet. |
| tableOps.setProperty(doctable, "table.bloom.key.functor", ColumnFamilyFunctor.class.getCanonicalName()); |
| tableOps.setProperty(doctable, "table.bloom.enabled", Boolean.TRUE.toString()); |
| } |
| |
| // Set mtbw by calling setMultiTableBatchWriter(). The DAO does this and manages flushing. |
| // If you create it here, tests work, but a real Accumulo may lose writes due to unmanaged flushing. |
| if (mtbw != null) { |
| docTableBw = mtbw.getBatchWriter(doctable); |
| termTableBw = mtbw.getBatchWriter(termtable); |
| } |
| tokenizer = ConfigUtils.getFreeTextTokenizer(conf); |
| validPredicates = ConfigUtils.getFreeTextPredicates(conf); |
| |
| queryTermLimit = ConfigUtils.getFreeTextTermLimit(conf); |
| } |
| |
| /** |
| * setConf sets the configuration and then initializes for query only. |
| * Use this alone if usage does not require writing. |
| * For a writeable (store and delete) version of this, |
| * call this and then setMultiTableBatchWriter(), then call init() |
| * that is what the DAO does. |
| */ |
| @Override |
| public void setConf(final Configuration conf) { |
| this.conf = conf; |
| if (!isInit) { |
| try { |
| initInternal(); |
| isInit = true; |
| } catch (AccumuloException | AccumuloSecurityException | TableNotFoundException | TableExistsException e) { |
| logger.warn("Unable to initialize index. Throwing Runtime Exception. ", e); |
| throw new RuntimeException(e); |
| } |
| } |
| } |
| |
| @Override |
| public Configuration getConf() { |
| return conf; |
| } |
| |
| |
| private void storeStatement(final Statement statement) throws IOException { |
| Objects.requireNonNull(mtbw, "Freetext indexer attempting to store, but setMultiTableBatchWriter() was not set."); |
| |
| // if the predicate list is empty, accept all predicates. |
| // Otherwise, make sure the predicate is on the "valid" list |
| final boolean isValidPredicate = validPredicates.isEmpty() || validPredicates.contains(statement.getPredicate()); |
| |
| if (isValidPredicate && (statement.getObject() instanceof Literal)) { |
| |
| // Get the tokens |
| final String text = statement.getObject().stringValue().toLowerCase(); |
| final SortedSet<String> tokens = tokenizer.tokenize(text); |
| |
| if (!tokens.isEmpty()) { |
| // Get Document Data |
| final String docContent = StatementSerializer.writeStatement(statement); |
| |
| final String docId = Md5Hash.md5Base64(docContent); |
| |
| // Setup partition |
| final Text partition = genPartition(docContent.hashCode(), docTableNumPartitions); |
| |
| final Mutation docTableMut = new Mutation(partition); |
| final List<Mutation> termTableMutations = new ArrayList<Mutation>(); |
| |
| final Text docIdText = new Text(docId); |
| |
| // Store the Document Data |
| docTableMut.put(ColumnPrefixes.DOCS_CF_PREFIX, docIdText, new Value(docContent.getBytes(Charsets.UTF_8))); |
| |
| // index the statement parts |
| docTableMut.put(ColumnPrefixes.getSubjColFam(statement), docIdText, EMPTY_VALUE); |
| docTableMut.put(ColumnPrefixes.getPredColFam(statement), docIdText, EMPTY_VALUE); |
| docTableMut.put(ColumnPrefixes.getObjColFam(statement), docIdText, EMPTY_VALUE); |
| docTableMut.put(ColumnPrefixes.getContextColFam(statement), docIdText, EMPTY_VALUE); |
| |
| // index the statement terms |
| for (final String token : tokens) { |
| // tie the token to the document |
| docTableMut.put(ColumnPrefixes.getTermColFam(token), docIdText, EMPTY_VALUE); |
| |
| // store the term in the term table (useful for wildcard searches) |
| termTableMutations.add(createEmptyPutMutation(ColumnPrefixes.getTermListColFam(token))); |
| termTableMutations.add(createEmptyPutMutation(ColumnPrefixes.getRevTermListColFam(token))); |
| } |
| |
| // write the mutations |
| try { |
| docTableBw.addMutation(docTableMut); |
| termTableBw.addMutations(termTableMutations); |
| } catch (final MutationsRejectedException e) { |
| logger.error("error adding mutation", e); |
| throw new IOException(e); |
| } |
| |
| } |
| |
| } |
| } |
| |
| @Override |
| public void storeStatement(final RyaStatement statement) throws IOException { |
| storeStatement(RyaToRdfConversions.convertStatement(statement)); |
| } |
| |
| private static Mutation createEmptyPutMutation(final Text row) { |
| final Mutation m = new Mutation(row); |
| m.put(EMPTY_TEXT, EMPTY_TEXT, EMPTY_VALUE); |
| return m; |
| } |
| |
| private static Mutation createEmptyPutDeleteMutation(final Text row) { |
| final Mutation m = new Mutation(row); |
| m.putDelete(EMPTY_TEXT, EMPTY_TEXT); |
| return m; |
| } |
| |
| private static Text genPartition(final int partition, final int numParitions) { |
| final int length = Integer.toString(numParitions).length(); |
| return new Text(String.format("%0" + length + "d", Math.abs(partition % numParitions))); |
| } |
| |
| @Override |
| public Set<IRI> getIndexablePredicates() { |
| return validPredicates; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override |
| public void flush() throws IOException { |
| try { |
| mtbw.flush(); |
| } catch (final MutationsRejectedException e) { |
| logger.error("error flushing the batch writer", e); |
| throw new IOException(e); |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override |
| public void close() throws IOException { |
| try { |
| if (mtbw!=null) |
| mtbw.close(); |
| } catch (final MutationsRejectedException e) { |
| logger.error("error closing the batch writer", e); |
| throw new IOException(e); |
| } |
| } |
| |
| private Set<String> unrollWildcard(final String string, final boolean reverse) throws IOException { |
| final Scanner termTableScan = getScanner(getFreeTextTermTablename(conf)); |
| |
| final Set<String> unrolledTerms = new HashSet<String>(); |
| |
| Text queryTerm; |
| if (reverse) { |
| final String t = StringUtils.removeStart(string, "*").toLowerCase(); |
| queryTerm = ColumnPrefixes.getRevTermListColFam(t); |
| } else { |
| final String t = StringUtils.removeEnd(string, "*").toLowerCase(); |
| queryTerm = ColumnPrefixes.getTermListColFam(t); |
| } |
| |
| // perform query and read results |
| termTableScan.setRange(Range.prefix(queryTerm)); |
| |
| for (final Entry<Key, Value> e : termTableScan) { |
| final String term = ColumnPrefixes.removePrefix(e.getKey().getRow()).toString(); |
| if (reverse) { |
| unrolledTerms.add(StringUtils.reverse(term)); |
| } else { |
| unrolledTerms.add(term); |
| } |
| } |
| |
| if (unrolledTerms.isEmpty()) { |
| // put in a placeholder term that will never be in the index. |
| unrolledTerms.add("\1\1\1"); |
| } |
| |
| return unrolledTerms; |
| } |
| |
| private void unrollWildcards(final SimpleNode node) throws IOException { |
| if (node instanceof ASTExpression || node instanceof ASTSimpleNode) { |
| for (final SimpleNode n : getNodeIterator(node)) { |
| unrollWildcards(n); |
| } |
| } else if (node instanceof ASTTerm) { |
| final ASTTerm term = (ASTTerm) node; |
| final boolean isWildTerm = term.getType().equals(ASTTerm.WILDTERM); |
| final boolean isPreWildTerm = term.getType().equals(ASTTerm.PREFIXTERM); |
| if (isWildTerm || isPreWildTerm) { |
| final Set<String> unrolledTerms = unrollWildcard(term.getTerm(), isPreWildTerm); |
| |
| // create a new expression |
| final ASTExpression newExpression = new ASTExpression(QueryParserTreeConstants.JJTEXPRESSION); |
| newExpression.setType(ASTExpression.OR); |
| newExpression.setNotFlag(term.isNotFlag()); |
| |
| for (final String unrolledTerm : unrolledTerms) { |
| final ASTTerm t = new ASTTerm(QueryParserTreeConstants.JJTTERM); |
| t.setNotFlag(false); |
| t.setTerm(unrolledTerm); |
| t.setType(ASTTerm.TERM); |
| ASTNodeUtils.pushChild(newExpression, t); |
| } |
| |
| // replace "term" node with "expression" node in "term" node parent |
| final SimpleNode parent = (SimpleNode) term.jjtGetParent(); |
| final int index = ASTNodeUtils.getChildIndex(parent, term); |
| |
| Validate.isTrue(index >= 0, "child not found in parent"); |
| |
| parent.jjtAddChild(newExpression, index); |
| } |
| |
| } else { |
| throw new IllegalArgumentException("Node is of unknown type: " + node.getClass().getName()); |
| } |
| } |
| |
| private Scanner getScanner(final String tablename) throws IOException { |
| try { |
| return ConfigUtils.createScanner(tablename, conf); |
| } catch (AccumuloException | AccumuloSecurityException | TableNotFoundException e) { |
| logger.error("Error connecting to " + tablename); |
| throw new IOException(e); |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override |
| public CloseableIteration<Statement, QueryEvaluationException> queryText(final String query, final StatementConstraints contraints) |
| throws IOException { |
| final Scanner docTableScan = getScanner(getFreeTextDocTablename(conf)); |
| |
| // test the query to see if it's parses correctly. |
| SimpleNode root = parseQuery(query); |
| |
| // unroll any wildcard nodes before it goes to the server |
| unrollWildcards(root); |
| |
| final String unrolledQuery = ASTNodeUtils.serializeExpression(root); |
| |
| // Add S P O C constraints to query |
| final StringBuilder constrainedQuery = new StringBuilder("(" + unrolledQuery + ")"); |
| |
| if (contraints.hasSubject()) { |
| constrainedQuery.append(" AND "); |
| constrainedQuery.append(ColumnPrefixes.getSubjColFam(contraints.getSubject().toString()).toString()); |
| } |
| if (contraints.hasContext()) { |
| constrainedQuery.append(" AND "); |
| constrainedQuery.append(ColumnPrefixes.getContextColFam(contraints.getContext().toString()).toString()); |
| } |
| if (contraints.hasPredicates()) { |
| constrainedQuery.append(" AND ("); |
| final List<String> predicates = new ArrayList<String>(); |
| for (final IRI u : contraints.getPredicates()) { |
| predicates.add(ColumnPrefixes.getPredColFam(u.stringValue()).toString()); |
| } |
| constrainedQuery.append(StringUtils.join(predicates, " OR ")); |
| constrainedQuery.append(")"); |
| } |
| |
| // Verify that the query is a reasonable size |
| root = parseQuery(constrainedQuery.toString()); |
| final int termCount = ASTNodeUtils.termCount(root); |
| |
| if (termCount > queryTermLimit) { |
| throw new IOException("Query contains too many terms. Term limit: " + queryTermLimit + ". Term Count: " + termCount); |
| } |
| |
| // perform query |
| docTableScan.clearScanIterators(); |
| docTableScan.clearColumns(); |
| |
| final int iteratorPriority = 20; |
| final String iteratorName = "booleanTree"; |
| final IteratorSetting ii = new IteratorSetting(iteratorPriority, iteratorName, BooleanTreeIterator.class); |
| BooleanTreeIterator.setQuery(ii, constrainedQuery.toString()); |
| docTableScan.addScanIterator(ii); |
| docTableScan.setRange(new Range()); |
| |
| return getIteratorWrapper(docTableScan); |
| } |
| |
| private static CloseableIteration<Statement, QueryEvaluationException> getIteratorWrapper(final Scanner s) { |
| |
| final Iterator<Entry<Key, Value>> i = s.iterator(); |
| |
| return new CloseableIteration<Statement, QueryEvaluationException>() { |
| @Override |
| public boolean hasNext() { |
| return i.hasNext(); |
| } |
| |
| @Override |
| public Statement next() throws QueryEvaluationException { |
| final Entry<Key, Value> entry = i.next(); |
| final Value v = entry.getValue(); |
| try { |
| final String dataString = Text.decode(v.get(), 0, v.getSize()); |
| final Statement s = StatementSerializer.readStatement(dataString); |
| return s; |
| } catch (final CharacterCodingException e) { |
| logger.error("Error decoding value", e); |
| throw new QueryEvaluationException(e); |
| } catch (final IOException e) { |
| logger.error("Error deserializing statement", e); |
| throw new QueryEvaluationException(e); |
| } |
| } |
| |
| @Override |
| public void remove() { |
| throw new UnsupportedOperationException("Remove not implemented"); |
| } |
| |
| @Override |
| public void close() throws QueryEvaluationException { |
| if (s != null) { |
| s.close(); |
| } |
| } |
| }; |
| } |
| |
| /** |
| * Simple adapter that parses the query using {@link QueryParser}. Note: any checked exceptions thrown by {@link QueryParser} are |
| * re-thrown as {@link IOException}s. |
| * |
| * @param query |
| * @return |
| * @throws IOException |
| */ |
| private static SimpleNode parseQuery(final String query) throws IOException { |
| SimpleNode root = null; |
| try { |
| root = QueryParser.parse(query); |
| } catch (final ParseException e) { |
| logger.error("Parser Exception on Client Side. Query: " + query, e); |
| throw new IOException(e); |
| } catch (final TokenMgrError e) { |
| logger.error("Token Manager Exception on Client Side. Query: " + query, e); |
| throw new IOException(e); |
| } |
| return root; |
| } |
| |
| /** |
| * Get Free Text Document index table's name |
| * Use the two table version of this below. This one is required by base class. |
| */ |
| @Override |
| public String getTableName() { |
| return getFreeTextDocTablename(conf); |
| } |
| |
| /** |
| * Make the Accumulo table names used by this indexer for a specific instance of Rya. |
| * |
| * @param conf - The Rya configuration that specifies which instance of Rya |
| * the table names will be built for. (not null) |
| * @return The Accumulo table names used by this indexer for a specific instance of Rya. |
| */ |
| public static List<String> getTableNames(final Configuration conf) { |
| requireNonNull(conf); |
| return Collections.unmodifiableList( |
| makeTableNames(ConfigUtils.getTablePrefix(conf) ) ); |
| } |
| |
| /** |
| * Get the Document index's table name. |
| * |
| * @param conf - The Rya configuration that specifies which instance of Rya |
| * the table names will be built for. (not null) |
| * @return The Free Text Document index's Accumulo table name for the Rya instance. |
| */ |
| public static String getFreeTextDocTablename(final Configuration conf) { |
| requireNonNull(conf); |
| return makeFreeTextDocTablename( ConfigUtils.getTablePrefix(conf) ); |
| } |
| |
| @Override |
| public void setMultiTableBatchWriter(MultiTableBatchWriter writer) throws IOException { |
| mtbw = writer; |
| } |
| |
| |
| /** |
| * Get the Term index's table name. |
| * |
| * @param conf - The Rya configuration that specifies which instance of Rya |
| * the table names will be built for. (not null) |
| * @return The Free Text Term index's Accumulo table name for the Rya instance. |
| */ |
| public static String getFreeTextTermTablename(final Configuration conf) { |
| requireNonNull(conf); |
| return makeFreeTextTermTablename( ConfigUtils.getTablePrefix(conf) ); |
| } |
| |
| /** |
| * Make the Accumulo table names used by this indexer for a specific instance of Rya. |
| * |
| * @param ryaInstanceName - The name of the Rya instance the table names are for. (not null) |
| * @return The Accumulo table names used by this indexer for a specific instance of Rya. |
| */ |
| public static List<String> makeTableNames(final String ryaInstanceName) { |
| requireNonNull(ryaInstanceName); |
| return Lists.newArrayList( |
| makeFreeTextDocTablename(ryaInstanceName), |
| makeFreeTextTermTablename(ryaInstanceName)); |
| } |
| |
| /** |
| * Make the Document index's table name. |
| * |
| * @param ryaInstanceName - The name of the Rya instance the table names are for. (not null) |
| * @return The Free Text Document index's Accumulo table name for the Rya instance. |
| */ |
| public static String makeFreeTextDocTablename(final String ryaInstanceName) { |
| requireNonNull(ryaInstanceName); |
| return ryaInstanceName + TABLE_SUFFFIX_DOC; |
| } |
| |
| /** |
| * Make the Term index's table name. |
| * |
| * @param ryaInstanceName - The name of the Rya instance the table names are for. (not null) |
| * @return The Free Text Term index's Accumulo table name for the Rya instance. |
| */ |
| public static String makeFreeTextTermTablename(final String ryaInstanceName) { |
| requireNonNull(ryaInstanceName); |
| return ryaInstanceName + TABLE_SUFFIX_TERM; |
| } |
| |
| private void deleteStatement(final Statement statement) throws IOException { |
| Objects.requireNonNull(mtbw, "Freetext indexer attempting to delete, but setMultiTableBatchWriter() was not set."); |
| |
| // if the predicate list is empty, accept all predicates. |
| // Otherwise, make sure the predicate is on the "valid" list |
| final boolean isValidPredicate = validPredicates.isEmpty() || validPredicates.contains(statement.getPredicate()); |
| |
| if (isValidPredicate && (statement.getObject() instanceof Literal)) { |
| |
| // Get the tokens |
| final String text = statement.getObject().stringValue().toLowerCase(); |
| final SortedSet<String> tokens = tokenizer.tokenize(text); |
| |
| if (!tokens.isEmpty()) { |
| // Get Document Data |
| final String docContent = StatementSerializer.writeStatement(statement); |
| |
| final String docId = Md5Hash.md5Base64(docContent); |
| |
| // Setup partition |
| final Text partition = genPartition(docContent.hashCode(), docTableNumPartitions); |
| |
| final Mutation docTableMut = new Mutation(partition); |
| final List<Mutation> termTableMutations = new ArrayList<Mutation>(); |
| |
| final Text docIdText = new Text(docId); |
| |
| // Delete the Document Data |
| docTableMut.putDelete(ColumnPrefixes.DOCS_CF_PREFIX, docIdText); |
| |
| // Delete the statement parts in index |
| docTableMut.putDelete(ColumnPrefixes.getSubjColFam(statement), docIdText); |
| docTableMut.putDelete(ColumnPrefixes.getPredColFam(statement), docIdText); |
| docTableMut.putDelete(ColumnPrefixes.getObjColFam(statement), docIdText); |
| docTableMut.putDelete(ColumnPrefixes.getContextColFam(statement), docIdText); |
| |
| |
| // Delete the statement terms in index |
| for (final String token : tokens) { |
| if (IS_TERM_TABLE_TOKEN_DELETION_ENABLED) { |
| final int rowId = Integer.parseInt(partition.toString()); |
| final boolean doesTermExistInOtherDocs = doesTermExistInOtherDocs(token, rowId, docIdText); |
| // Only delete the term from the term table if it doesn't appear in other docs |
| if (!doesTermExistInOtherDocs) { |
| // Delete the term in the term table |
| termTableMutations.add(createEmptyPutDeleteMutation(ColumnPrefixes.getTermListColFam(token))); |
| termTableMutations.add(createEmptyPutDeleteMutation(ColumnPrefixes.getRevTermListColFam(token))); |
| } |
| } |
| |
| // Un-tie the token to the document |
| docTableMut.putDelete(ColumnPrefixes.getTermColFam(token), docIdText); |
| } |
| |
| // write the mutations |
| try { |
| docTableBw.addMutation(docTableMut); |
| termTableBw.addMutations(termTableMutations); |
| } catch (final MutationsRejectedException e) { |
| logger.error("error adding mutation", e); |
| throw new IOException(e); |
| } |
| |
| } |
| } |
| } |
| |
| @Override |
| public void deleteStatement(final RyaStatement statement) throws IOException { |
| deleteStatement(RyaToRdfConversions.convertStatement(statement)); |
| } |
| |
| /** |
| * Checks to see if the provided term appears in other documents. |
| * @param term the term to search for. |
| * @param currentDocId the current document ID that the search term exists in. |
| * @return {@code true} if the term was found in other documents. {@code false} otherwise. |
| */ |
| private boolean doesTermExistInOtherDocs(final String term, final int currentDocId, final Text docIdText) { |
| try { |
| final String freeTextDocTableName = getFreeTextDocTablename(conf); |
| final Scanner scanner = getScanner(freeTextDocTableName); |
| |
| final String t = StringUtils.removeEnd(term, "*").toLowerCase(); |
| final Text queryTerm = ColumnPrefixes.getTermColFam(t); |
| |
| // perform query and read results |
| scanner.fetchColumnFamily(queryTerm); |
| |
| for (final Entry<Key, Value> entry : scanner) { |
| final Key key = entry.getKey(); |
| final Text row = key.getRow(); |
| final int rowId = Integer.parseInt(row.toString()); |
| // We only want to check other documents from the one we're deleting |
| if (rowId != currentDocId) { |
| final Text columnFamily = key.getColumnFamily(); |
| final String columnFamilyValue = columnFamily.toString(); |
| // Check that the value has the term prefix |
| if (columnFamilyValue.startsWith(ColumnPrefixes.TERM_CF_PREFIX.toString())) { |
| final Text text = ColumnPrefixes.removePrefix(columnFamily); |
| final String value = text.toString(); |
| if (value.equals(term)) { |
| return true; |
| } |
| } |
| } |
| } |
| } catch (final IOException e) { |
| logger.error("Error searching for the existance of the term in other documents", e); |
| } |
| return false; |
| } |
| |
| |
| /** |
| * called by the DAO after setting the mtbw. |
| * The rest of the initilization is done by setConf() |
| */ |
| @Override |
| public void init() { |
| Objects.requireNonNull(mtbw, "Freetext indexer failed to initialize temporal index, setMultiTableBatchWriter() was not set."); |
| Objects.requireNonNull(conf, "Freetext indexer failed to initialize temporal index, setConf() was not set."); |
| try { |
| docTableBw = mtbw.getBatchWriter(getFreeTextDocTablename(conf)); |
| termTableBw = mtbw.getBatchWriter(getFreeTextTermTablename(conf)); |
| } catch (AccumuloException | AccumuloSecurityException | TableNotFoundException e) { |
| logger.error("Unable to initialize index. Throwing Runtime Exception. ", e); |
| throw new RuntimeException(e); |
| } |
| } |
| |
| |
| @Override |
| public void setConnector(final Connector connector) { |
| // TODO Auto-generated method stub |
| |
| } |
| |
| |
| @Override |
| public void destroy() { |
| // TODO Auto-generated method stub |
| |
| } |
| |
| |
| @Override |
| public void purge(final RdfCloudTripleStoreConfiguration configuration) { |
| // TODO Auto-generated method stub |
| |
| } |
| |
| |
| @Override |
| public void dropAndDestroy() { |
| // TODO Auto-generated method stub |
| |
| } |
| } |
