nativerdfrepository.java

这是外国一个开源推理机

/*  Sesame - Storage and Querying architecture for RDF and RDF Schema
 *  Copyright (C) 2001-2005 Aduna
 *
 *  Contact:
 *  	Aduna
 *  	Prinses Julianaplein 14 b
 *  	3817 CS Amersfoort
 *  	The Netherlands
 *  	tel. [phone]
 *  	fax. [phone]
 *
 *  	http://aduna.biz/
 *  	http://www.openrdf.org/
 *
 *  This library is free software; you can redistribute it and/or
 *  modify it under the terms of the GNU Lesser General Public
 *  License as published by the Free Software Foundation; either
 *  version 2.1 of the License, or (at your option) any later version.
 *
 *  This library is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 *  Lesser General Public License for more details.
 *
 *  You should have received a copy of the GNU Lesser General Public
 *  License along with this library; if not, write to the Free Software
 *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */

package org.openrdf.sesame.sailimpl.nativerdf;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.openrdf.util.log.ThreadLog;

import org.openrdf.model.BNode;
import org.openrdf.model.Literal;
import org.openrdf.model.Resource;
import org.openrdf.model.URI;
import org.openrdf.model.Value;
import org.openrdf.model.ValueFactory;

import org.openrdf.rio.Parser;
import org.openrdf.rio.StatementHandler;
import org.openrdf.rio.StatementHandlerException;
import org.openrdf.rio.rdfxml.RdfXmlParser;

import org.openrdf.sesame.sail.NamespaceIterator;
import org.openrdf.sesame.sail.RdfRepository;
import org.openrdf.sesame.sail.SailChangedListener;
import org.openrdf.sesame.sail.SailInitializationException;
import org.openrdf.sesame.sail.SailInternalException;
import org.openrdf.sesame.sail.SailUpdateException;
import org.openrdf.sesame.sail.StatementIterator;
import org.openrdf.sesame.sail.query.Query;
import org.openrdf.sesame.sail.query.QueryOptimizer;
import org.openrdf.sesame.sail.query.Var;
import org.openrdf.sesame.sail.util.EmptyStatementIterator;
import org.openrdf.sesame.sail.util.SailChangedEventImpl;
import org.openrdf.sesame.sailimpl.nativerdf.model.NativeBNode;
import org.openrdf.sesame.sailimpl.nativerdf.model.NativeLiteral;
import org.openrdf.sesame.sailimpl.nativerdf.model.NativeURI;

/**
 * An implementation of the RdfRepository interface from the RDF Sail API that
 * stores its data in, and queries it from files on disk.
 * 
 * @author Arjohn Kampman
 */
public class NativeRdfRepository implements RdfRepository {

	/*--------------+
	 | Constants     |
	 +--------------*/

	/** Key used to specify a data directory in the initialization parameters. **/
	public static final String DATA_DIR_KEY = "dir";

	/** Key used to specify which triple indexes to use. **/
	public static final String TRIPLES_INDEXES_KEY = "triple-indexes";

	/*--------------+
	 | Variables     |
	 +--------------*/

	private File _dataDir;

	private TripleStore _tripleStore;

	private ValueStore _valueStore;

	private NamespaceStore _namespaceStore;

	private ValueFactory _valueFactory;

	protected List _sailChangedListeners;

	/**
	 * Flag indicating whether a transaction has been started.
	 **/
	protected boolean _transactionStarted;

	protected SailChangedEventImpl _sailChangedEvent;

	/*--------------+
	 | Constructors  |
	 +--------------*/

	/**
	 * Creates a new NativeRdfRepository.
	 **/
	public NativeRdfRepository() {
		_transactionStarted = false;

		_sailChangedListeners = new ArrayList(0);
	}

	/*--------------+
	 | Methods       |
	 +--------------*/

	/**
	 * Initializes this NativeRdfRepository. The supplied Map can contain the following
	 * parameters:
	 * <table>
	 * <th><td>key</td><td>value</td></th>
	 * <tr><td>dir</td><td>The fully qualified name of the data directory</td></tr>
	 * </table>
	 * 
	 * @param configParams The configuration parameters.
	 * @exception SailInitializationException If this RdfRepository could not be
	 * initialized using the supplied parameters.
	 * @see #DATA_DIR_KEY
	 **/
	public void initialize(Map configParams)
		throws SailInitializationException
	{
		// Get initialization parameters
		String dir = (String)configParams.get(DATA_DIR_KEY);
		if (dir == null) {
			throw new SailInitializationException("Missing parameter: dir");
		}

		String tripleIndexes = (String)configParams.get(TRIPLES_INDEXES_KEY);

		initialize(new File(dir), tripleIndexes);
	}

	/**
	 * Initializes this repository.
	 * 
	 * @param dataDir The data directory.
	 * @exception SailInternalException If the initialization failed.
	 */
	public void initialize(File dataDir)
		throws SailInitializationException
	{
		initialize(dataDir, null);
	}

	/**
	 * Initializes this repository.
	 * 
	 * @param dataDir The data directory.
	 * @exception SailInternalException If the initialization failed.
	 */
	public void initialize(File dataDir, String tripleIndexes)
		throws SailInitializationException
	{
		ThreadLog.trace("Initializing native RDF repository...");

		_dataDir = dataDir;
		if (!_dataDir.exists()) {
			boolean success = _dataDir.mkdirs();
			if (!success) {
				throw new SailInitializationException(
						"Unable to create data directory: " + dataDir);
			}
		}
		else if (!_dataDir.isDirectory()) {
			throw new SailInitializationException(
					"The specified path does not denote a directory: " + dataDir);
		}
		else if (!_dataDir.canRead()) {
			throw new SailInitializationException(
					"Not allowed to read from the specified directory: " + dataDir);
		}
		else if (!_dataDir.canWrite()) {
			throw new SailInitializationException(
					"Not allowed to write to the specified directory: " + dataDir);
		}

		try {
			_namespaceStore = new NamespaceStore(_dataDir);
			_valueFactory = _valueStore = new ValueStore(_dataDir, _namespaceStore, this);
			_tripleStore = new TripleStore(_dataDir, tripleIndexes);
		}
		catch (IOException e) {
			throw new SailInitializationException(e);
		}

		ThreadLog.trace("native RDF repository initialized");
	}

	// Implements Sail.shutDown()
	public void shutDown() {
		try {
			_tripleStore.close();
			_valueStore.close();
		}
		catch (IOException e) {
			throw new SailInternalException(e);
		}
	}

	// Implements RdfSource.getValueFactory()
	public ValueFactory getValueFactory() {
		return _valueFactory;
	}

	// Implements RdfSource.getStatements(Resource, URI, Value)
	public StatementIterator getStatements(Resource subj, URI pred, Value obj) {
		try {
			int subjID = 0;
			if (subj != null) {
				subjID = _valueStore.getID(subj);
				if (subjID == 0) {
					return new EmptyStatementIterator();
				}
			}

			int predID = 0;
			if (pred != null) {
				predID = _valueStore.getID(pred);
				if (predID == 0) {
					return new EmptyStatementIterator();
				}
			}

			int objID = 0;
			if (obj != null) {
				objID = _valueStore.getID(obj);
				if (objID == 0) {
					return new EmptyStatementIterator();
				}
			}

			return new NativeStatementIterator(_tripleStore, _valueStore,
					_valueFactory, subj, pred, obj, subjID, predID, objID);
		}
		catch (IOException e) {
			throw new SailInternalException(e);
		}
	}

	// Implements RdfSource.hasStatement(Resource, URI, Value)
	public boolean hasStatement(Resource subj, URI pred, Value obj) {
		StatementIterator stIter = getStatements(subj, pred, obj);
		boolean result = stIter.hasNext();
		stIter.close();
		return result;
	}

	// Implements RdfSource.optimizeQuery(Query)
	public Query optimizeQuery(Query qc) {
		// Apply the default optimizations
		QueryOptimizer.optimizeQuery(qc);

		// Replace all Value objects stored in variables with NativeValue objects
		_replaceValuesInQuery(qc);
		return qc;
	}

	/**
	 * Replaces all Value objects stored in variables with NativeValue objects.
	 **/
	private void _replaceValuesInQuery(Query query) {
		List varList = new ArrayList();
		query.getVariables(varList);

		for (int i = 0; i < varList.size(); i++) {
			Var var = (Var)varList.get(i);

			if (var.hasValue()) {
				Value value = var.getValue();
				if (value instanceof URI) {
					var.setValue(new NativeURI(this, (URI)value));
				}
				else if (value instanceof BNode) {
					var.setValue(new NativeBNode(this, (BNode)value));
				}
				else if (value instanceof Literal) {
					Literal lit = (Literal)value;
					if (lit.getLanguage() != null) {
						lit = new NativeLiteral(this, lit.getLabel(),
								lit.getLanguage());
					}
					else if (lit.getDatatype() != null) {
						lit = new NativeLiteral(this, lit.getLabel(),
								lit.getDatatype());
					}
					else {
						lit = new NativeLiteral(this, lit.getLabel());
					}
					var.setValue(lit);
				}
				// else: do not change the value
			}
		}
	}

	// Implements RdfSource.getNamespaces()
	public NamespaceIterator getNamespaces() {
		return _namespaceStore.getNamespaces();
	}

	public void startTransaction() {
		try {
			_valueStore.startTransaction();
			_tripleStore.startTransaction();
			_transactionStarted = true;
			_sailChangedEvent = new SailChangedEventImpl();
		}
		catch (IOException e) {
			throw new SailInternalException(e);
		}
	}

	public void commitTransaction() {
		try {
			_tripleStore.commitTransaction();
			_valueStore.commitTransaction();
			_notifySailChanged(_sailChangedEvent);
		}
		catch (IOException e) {
			throw new SailInternalException(e);
		}
		finally {
			_sailChangedEvent = null;
			_transactionStarted = false;
		}
	}

	public boolean transactionStarted() {
		return _transactionStarted;
	}

	public void addStatement(Resource subj, URI pred, Value obj)
		throws SailUpdateException
	{
		if (!transactionStarted()) {
			throw new SailUpdateException("no transaction started.");
		}

		// FIXME: do some batch-wise processing, e.g. per 1000 statements?

		try {
			int subjID = _valueStore.storeValue(subj);
			int predID = _valueStore.storeValue(pred);
			int objID = _valueStore.storeValue(obj);

			byte[] oldValue = _tripleStore.storeTriple(subjID, predID, objID);

			if (oldValue != null) {
				// The triple replaced another triple with identical subject, predicate
				// and object. Decrease the reference counts for these values
				_valueStore.changeReferenceCount(subjID, -1);
				_valueStore.changeReferenceCount(predID, -1);
				_valueStore.changeReferenceCount(objID, -1);
			}
			else {
				_sailChangedEvent.setStatementsAdded(true);
			}
		}
		catch (IOException e) {
			throw new SailUpdateException(e);
		}
	}

	public int removeStatements(Resource subj, URI pred, Value obj)
		throws SailUpdateException
	{
		if (!transactionStarted()) {
			throw new SailUpdateException("no transaction started.");
		}

		try {
			int subjID = 0;
			if (subj != null) {
				subjID = _valueStore.getID(subj);
				if (subjID == 0) {
					return 0;
				}
			}

			int predID = 0;
			if (pred != null) {
				predID = _valueStore.getID(pred);
				if (predID == 0) {
					return 0;
				}
			}

			int objID = 0;
			if (obj != null) {
				objID = _valueStore.getID(obj);
				if (objID == 0) {
					return 0;
				}
			}

			int count = _tripleStore.removeTriples(subjID, predID, objID);

			if (count > 0) {
				_sailChangedEvent.setStatementsRemoved(true);
			}

			// FIXME: this doesn't update the reference count correctly for all matches triples
			_valueStore.changeReferenceCount(subjID, -1);
			_valueStore.changeReferenceCount(predID, -1);
			_valueStore.changeReferenceCount(objID, -1);
			return count;
		}
		catch (IOException e) {
			throw new SailUpdateException(e);
		}
	}

	public void clearRepository()
		throws SailUpdateException
	{
		if (!transactionStarted()) {
			throw new SailUpdateException("no transaction started.");
		}

		try {
			_tripleStore.clear();
			_valueStore.clear();
			_sailChangedEvent.setStatementsRemoved(true);
		}
		catch (IOException e) {
			throw new SailUpdateException(e);
		}
	}

	public void changeNamespacePrefix(String namespace, String prefix) {
		_namespaceStore.setNamespacePrefix(prefix, namespace);
	}

	public static void main(String[] args)
		throws Exception
	{
		File dataDir = new File(args[0]);
		File rdfFile = new File(args[1]);
		String baseURI = "foo:bar";
		if (args.length >= 3) {
			baseURI = args[2];
		}

		final NativeRdfRepository repository = new NativeRdfRepository();
		repository.initialize(dataDir);

		Parser parser = new RdfXmlParser(repository.getValueFactory());
		parser.setDatatypeHandling(Parser.DT_IGNORE);

		final InputStream inputStream = new FileInputStream(rdfFile);

		parser.setStatementHandler(new StatementHandler() {
			public void handleStatement(Resource subj, URI pred, Value obj)
				throws StatementHandlerException
			{
				try {
					repository.addStatement(subj, pred, obj);
				}
				catch (SailUpdateException e) {
					throw new StatementHandlerException(e);
				}
			}
		});

		long startTime = System.currentTimeMillis();

		repository.startTransaction();
		parser.parse(inputStream, baseURI);
		repository.commitTransaction();

		long endTime = System.currentTimeMillis();

		inputStream.close();

		System.out.println("Upload completed in " + (endTime - startTime) + " ms");
	}

	/* (non-Javadoc)
	 * @see org.openrdf.sesame.sail.RdfRepository#addListener(org.openrdf.sesame.sail.SailChangedListener)
	 */
	public void addListener(SailChangedListener listener) {
		synchronized(_sailChangedListeners) {
			_sailChangedListeners.add(listener);
		} // end synchronized block
	}

	/* (non-Javadoc)
	 * @see org.openrdf.sesame.sail.RdfRepository#removeListener(org.openrdf.sesame.sail.SailChangedListener)
	 */
	public void removeListener(SailChangedListener listener) {
		synchronized(_sailChangedListeners) {
			_sailChangedListeners.remove(listener);
		} // end synchronized block
	}

	protected void _notifySailChanged(SailChangedEventImpl event) {
		synchronized(_sailChangedListeners) {
			if (_sailChangedListeners != null) {
				if (event.sailChanged()) { // only notify if something actually changed
					Iterator listeners = _sailChangedListeners.iterator();

					while (listeners.hasNext()) {
						SailChangedListener listener = (SailChangedListener)listeners.next();
						listener.sailChanged(event);
					}
				}
			}
		} // end synchronized block
	}
}