| /* |
| * 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.accumulo; |
| |
| // |
| //import com.google.common.collect.Iterators; |
| //import com.google.common.io.ByteArrayDataInput; |
| //import com.google.common.io.ByteStreams; |
| //import org.eclipse.rdf4j.common.iteration.CloseableIteration; |
| //import org.apache.rya.api.RdfCloudTripleStoreConstants; |
| //import org.apache.rya.api.RdfCloudTripleStoreUtils; |
| //import org.apache.rya.api.persist.RdfDAOException; |
| //import org.apache.rya.api.utils.NullableStatementImpl; |
| //import org.apache.accumulo.core.client.*; |
| //import org.apache.accumulo.core.data.Key; |
| //import org.apache.accumulo.core.data.Range; |
| //import org.apache.accumulo.core.iterators.user.AgeOffFilter; |
| //import org.apache.accumulo.core.iterators.user.TimestampFilter; |
| //import org.apache.accumulo.core.security.Authorizations; |
| //import org.apache.hadoop.io.Text; |
| //import org.eclipse.rdf4j.model.Resource; |
| //import org.eclipse.rdf4j.model.Statement; |
| //import org.eclipse.rdf4j.model.IRI; |
| //import org.eclipse.rdf4j.model.Value; |
| //import org.eclipse.rdf4j.query.BindingSet; |
| //import org.slf4j.Logger; |
| //import org.slf4j.LoggerFactory; |
| // |
| //import java.io.IOException; |
| //import java.util.Collection; |
| //import java.util.Collections; |
| //import java.util.HashSet; |
| //import java.util.Iterator; |
| //import java.util.Map.Entry; |
| // |
| //import static org.apache.rya.accumulo.AccumuloRdfConstants.ALL_AUTHORIZATIONS; |
| //import static org.apache.rya.api.RdfCloudTripleStoreConstants.TABLE_LAYOUT; |
| //import static org.apache.rya.api.RdfCloudTripleStoreUtils.writeValue; |
| // |
| //public class AccumuloRdfQueryIterator implements |
| // CloseableIteration<Entry<Statement, BindingSet>, RdfDAOException> { |
| // |
| // protected final Logger logger = LoggerFactory.getLogger(getClass()); |
| // |
| // private boolean open = false; |
| // private Iterator result; |
| // private Resource[] contexts; |
| // private Collection<Entry<Statement, BindingSet>> statements; |
| // private int numOfThreads = 20; |
| // |
| // private RangeBindingSetEntries rangeMap = new RangeBindingSetEntries(); |
| // private ScannerBase scanner; |
| // private boolean isBatchScanner = true; |
| // private Statement statement; |
| // Iterator<BindingSet> iter_bss = null; |
| // |
| // private boolean hasNext = true; |
| // private AccumuloRdfConfiguration conf; |
| // private TABLE_LAYOUT tableLayout; |
| // private Text context_txt; |
| // |
| // private DefineTripleQueryRangeFactory queryRangeFactory = new DefineTripleQueryRangeFactory(); |
| // |
| // public AccumuloRdfQueryIterator(Collection<Entry<Statement, BindingSet>> statements, Connector connector, Resource... contexts) |
| // throws RdfDAOException { |
| // this(statements, connector, null, contexts); |
| // } |
| // |
| // public AccumuloRdfQueryIterator(Collection<Entry<Statement, BindingSet>> statements, Connector connector, |
| // AccumuloRdfConfiguration conf, Resource... contexts) |
| // throws RdfDAOException { |
| // this.statements = statements; |
| // this.contexts = contexts; |
| // this.conf = conf; |
| // initialize(connector); |
| // open = true; |
| // } |
| // |
| // public AccumuloRdfQueryIterator(Resource subject, URI predicate, Value object, Connector connector, |
| // AccumuloRdfConfiguration conf, Resource[] contexts) throws RdfDAOException { |
| // this(Collections.<Entry<Statement, BindingSet>>singleton(new RdfCloudTripleStoreUtils.CustomEntry<Statement, BindingSet>( |
| // new NullableStatementImpl(subject, predicate, object, contexts), |
| // null)), connector, conf, contexts); |
| // } |
| // |
| // protected void initialize(Connector connector) |
| // throws RdfDAOException { |
| // try { |
| // //TODO: We cannot span multiple tables here |
| // Collection<Range> ranges = new HashSet<Range>(); |
| // |
| // result = Iterators.emptyIterator(); |
| // Long startTime = conf.getStartTime(); |
| // Long ttl = conf.getTtl(); |
| // |
| // Resource context = null; |
| // for (Entry<Statement, BindingSet> stmtbs : statements) { |
| // Statement stmt = stmtbs.getKey(); |
| // Resource subject = stmt.getSubject(); |
| // URI predicate = stmt.getPredicate(); |
| // Value object = stmt.getObject(); |
| // context = stmt.getContext(); //TODO: assumes the same context for all statements |
| // logger.debug("Batch Scan, lookup subject[" + subject + "] predicate[" + predicate + "] object[" + object + "] combination"); |
| // |
| // Entry<TABLE_LAYOUT, Range> entry = queryRangeFactory.defineRange(subject, predicate, object, conf); |
| // tableLayout = entry.getKey(); |
| //// isTimeRange = isTimeRange || queryRangeFactory.isTimeRange(); |
| // Range range = entry.getValue(); |
| // ranges.add(range); |
| // rangeMap.ranges.add(new RdfCloudTripleStoreUtils.CustomEntry<Range, BindingSet>(range, stmtbs.getValue())); |
| // } |
| // |
| // Authorizations authorizations = AccumuloRdfConstants.ALL_AUTHORIZATIONS; |
| // String auth = conf.getAuth(); |
| // if (auth != null) { |
| // authorizations = new Authorizations(auth.split(",")); |
| // } |
| // String table = RdfCloudTripleStoreUtils.layoutToTable(tableLayout, conf); |
| // result = createScanner(connector, authorizations, table, context, startTime, ttl, ranges); |
| //// if (isBatchScanner) { |
| //// ((BatchScanner) scanner).setRanges(ranges); |
| //// } else { |
| //// for (Range range : ranges) { |
| //// ((Scanner) scanner).setRange(range); //TODO: Not good way of doing this |
| //// } |
| //// } |
| //// |
| //// if (isBatchScanner) { |
| //// result = ((BatchScanner) scanner).iterator(); |
| //// } else { |
| //// result = ((Scanner) scanner).iterator(); |
| //// } |
| // } catch (Exception e) { |
| // throw new RdfDAOException(e); |
| // } |
| // } |
| // |
| // protected Iterator<Entry<Key, org.apache.accumulo.core.data.Value>> createScanner(Connector connector, Authorizations authorizations, String table, Resource context, Long startTime, Long ttl, Collection<Range> ranges) throws TableNotFoundException, IOException { |
| //// ShardedConnector shardedConnector = new ShardedConnector(connector, 4, ta) |
| // if (rangeMap.ranges.size() > (numOfThreads / 2)) { //TODO: Arbitrary number, make configurable |
| // BatchScanner scannerBase = connector.createBatchScanner(table, authorizations, numOfThreads); |
| // scannerBase.setRanges(ranges); |
| // populateScanner(context, startTime, ttl, scannerBase); |
| // return scannerBase.iterator(); |
| // } else { |
| // isBatchScanner = false; |
| // Iterator<Entry<Key, org.apache.accumulo.core.data.Value>>[] iters = new Iterator[ranges.size()]; |
| // int i = 0; |
| // for (Range range : ranges) { |
| // Scanner scannerBase = connector.createScanner(table, authorizations); |
| // populateScanner(context, startTime, ttl, scannerBase); |
| // scannerBase.setRange(range); |
| // iters[i] = scannerBase.iterator(); |
| // i++; |
| // scanner = scannerBase; //TODO: Always overridden, but doesn't matter since Scanner doesn't need to be closed |
| // } |
| // return Iterators.concat(iters); |
| // } |
| // |
| // } |
| // |
| // protected void populateScanner(Resource context, Long startTime, Long ttl, ScannerBase scannerBase) throws IOException { |
| // if (context != null) { //default graph |
| // context_txt = new Text(writeValue(context)); |
| // scannerBase.fetchColumnFamily(context_txt); |
| // } |
| // |
| //// if (!isQueryTimeBased(conf)) { |
| // if (startTime != null && ttl != null) { |
| //// scannerBase.setScanIterators(1, FilteringIterator.class.getName(), "filteringIterator"); |
| //// scannerBase.setScanIteratorOption("filteringIterator", "0", TimeRangeFilter.class.getName()); |
| //// scannerBase.setScanIteratorOption("filteringIterator", "0." + TimeRangeFilter.TIME_RANGE_PROP, ttl); |
| //// scannerBase.setScanIteratorOption("filteringIterator", "0." + TimeRangeFilter.START_TIME_PROP, startTime); |
| // IteratorSetting setting = new IteratorSetting(1, "fi", TimestampFilter.class.getName()); |
| // TimestampFilter.setStart(setting, startTime, true); |
| // TimestampFilter.setEnd(setting, startTime + ttl, true); |
| // scannerBase.addScanIterator(setting); |
| // } else if (ttl != null) { |
| //// scannerBase.setScanIterators(1, FilteringIterator.class.getName(), "filteringIterator"); |
| //// scannerBase.setScanIteratorOption("filteringIterator", "0", AgeOffFilter.class.getName()); |
| //// scannerBase.setScanIteratorOption("filteringIterator", "0.ttl", ttl); |
| // IteratorSetting setting = new IteratorSetting(1, "fi", AgeOffFilter.class.getName()); |
| // AgeOffFilter.setTTL(setting, ttl); |
| // scannerBase.addScanIterator(setting); |
| // } |
| //// } |
| // } |
| // |
| // @Override |
| // public void close() throws RdfDAOException { |
| // if (!open) |
| // return; |
| // verifyIsOpen(); |
| // open = false; |
| // if (scanner != null && isBatchScanner) { |
| // ((BatchScanner) scanner).close(); |
| // } |
| // } |
| // |
| // public void verifyIsOpen() throws RdfDAOException { |
| // if (!open) { |
| // throw new RdfDAOException("Iterator not open"); |
| // } |
| // } |
| // |
| // @Override |
| // public boolean hasNext() throws RdfDAOException { |
| // try { |
| // if (!open) |
| // return false; |
| // verifyIsOpen(); |
| // /** |
| // * For some reason, the result.hasNext returns false |
| // * once at the end of the iterator, and then true |
| // * for every subsequent call. |
| // */ |
| // hasNext = (hasNext && result.hasNext()); |
| // return hasNext || ((iter_bss != null) && iter_bss.hasNext()); |
| // } catch (Exception e) { |
| // throw new RdfDAOException(e); |
| // } |
| // } |
| // |
| // @Override |
| // public Entry<Statement, BindingSet> next() throws RdfDAOException { |
| // try { |
| // if (!this.hasNext()) |
| // return null; |
| // |
| // return getStatement(result, contexts); |
| // } catch (Exception e) { |
| // throw new RdfDAOException(e); |
| // } |
| // } |
| // |
| // public Entry<Statement, BindingSet> getStatement( |
| // Iterator<Entry<Key, org.apache.accumulo.core.data.Value>> rowResults, |
| // Resource... filterContexts) throws IOException { |
| // try { |
| // while (true) { |
| // if (iter_bss != null && iter_bss.hasNext()) { |
| // return new RdfCloudTripleStoreUtils.CustomEntry<Statement, BindingSet>(statement, iter_bss.next()); |
| // } |
| // |
| // if (rowResults.hasNext()) { |
| // Entry<Key, org.apache.accumulo.core.data.Value> entry = rowResults.next(); |
| // Key key = entry.getKey(); |
| // ByteArrayDataInput input = ByteStreams.newDataInput(key.getRow().getBytes()); |
| // statement = RdfCloudTripleStoreUtils.translateStatementFromRow(input, key.getColumnFamily(), tableLayout, RdfCloudTripleStoreConstants.VALUE_FACTORY); |
| // iter_bss = rangeMap.containsKey(key).iterator(); |
| // } else |
| // break; |
| // } |
| // } catch (Exception e) { |
| // throw new IOException(e); |
| // } |
| // return null; |
| // } |
| // |
| // @Override |
| // public void remove() throws RdfDAOException { |
| // next(); |
| // } |
| // |
| // public int getNumOfThreads() { |
| // return numOfThreads; |
| // } |
| // |
| // public void setNumOfThreads(int numOfThreads) { |
| // this.numOfThreads = numOfThreads; |
| // } |
| // |
| // public DefineTripleQueryRangeFactory getQueryRangeFactory() { |
| // return queryRangeFactory; |
| // } |
| // |
| // public void setQueryRangeFactory(DefineTripleQueryRangeFactory queryRangeFactory) { |
| // this.queryRangeFactory = queryRangeFactory; |
| // } |
| //} |
