Menu
▾
▴
[Bigdata-commit] SF.net SVN: bigdata:[3750] branches/CHANGE_SET_BRANCH/bigdata
|
From: <mrp...@us...> - 2010-10-08 01:46:10
|
Revision: 3750
http://bigdata.svn.sourceforge.net/bigdata/?rev=3750&view=rev
Author: mrpersonick
Date: 2010-10-08 01:45:59 +0000 (Fri, 08 Oct 2010)
Log Message:
-----------
finishing up the change sets implementation
Modified Paths:
--------------
branches/CHANGE_SET_BRANCH/bigdata/src/java/com/bigdata/btree/proc/AbstractKeyArrayIndexProcedure.java
branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/inf/SPOAssertionBuffer.java
branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/inf/SPORetractionBuffer.java
branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/inf/TruthMaintenance.java
branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/model/BigdataStatementImpl.java
branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/rio/StatementBuffer.java
branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/spo/ISPO.java
branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/spo/SPO.java
branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/spo/SPOIndexRemover.java
branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/spo/SPOIndexWriteProc.java
branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/spo/SPOIndexWriter.java
branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/spo/StatementWriter.java
branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/store/AbstractTripleStore.java
branches/CHANGE_SET_BRANCH/bigdata-sails/src/java/com/bigdata/rdf/sail/BigdataSail.java
branches/CHANGE_SET_BRANCH/bigdata-sails/src/java/com/bigdata/rdf/sail/BigdataSailRepositoryConnection.java
branches/CHANGE_SET_BRANCH/bigdata-sails/src/test/com/bigdata/rdf/sail/TestChangeSets.java
Added Paths:
-----------
branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/changesets/
branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/changesets/ChangeRecord.java
branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/changesets/IChangeLog.java
branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/changesets/IChangeRecord.java
branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/changesets/StatementWriter.java
Removed Paths:
-------------
branches/CHANGE_SET_BRANCH/bigdata-sails/src/java/com/bigdata/rdf/sail/changesets/ChangeRecord.java
branches/CHANGE_SET_BRANCH/bigdata-sails/src/java/com/bigdata/rdf/sail/changesets/IChangeLog.java
branches/CHANGE_SET_BRANCH/bigdata-sails/src/java/com/bigdata/rdf/sail/changesets/IChangeRecord.java
branches/CHANGE_SET_BRANCH/bigdata-sails/src/java/com/bigdata/rdf/sail/changesets/StatementWriter.java
Modified: branches/CHANGE_SET_BRANCH/bigdata/src/java/com/bigdata/btree/proc/AbstractKeyArrayIndexProcedure.java
===================================================================
--- branches/CHANGE_SET_BRANCH/bigdata/src/java/com/bigdata/btree/proc/AbstractKeyArrayIndexProcedure.java 2010-10-07 20:23:05 UTC (rev 3749)
+++ branches/CHANGE_SET_BRANCH/bigdata/src/java/com/bigdata/btree/proc/AbstractKeyArrayIndexProcedure.java 2010-10-08 01:45:59 UTC (rev 3750)
@@ -39,6 +39,7 @@
import java.io.ObjectInput;
import java.io.ObjectOutput;
import java.io.OutputStream;
+import java.util.Arrays;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
@@ -795,18 +796,34 @@
IResultHandler<ResultBitBuffer, ResultBitBuffer> {
private final boolean[] results;
+
+ /**
+ * I added this so I could encode information about tuple modification
+ * that takes more than one boolean to encode. For example, SPOs can
+ * be: INSERTED, REMOVED, UPDATED, NO_OP (2 bits).
+ */
+ private final int multiplier;
+
private final AtomicInteger onCount = new AtomicInteger();
public ResultBitBufferHandler(final int nkeys) {
+
+ this(nkeys, 1);
+
+ }
+
+ public ResultBitBufferHandler(final int nkeys, final int multiplier) {
- results = new boolean[nkeys];
+ results = new boolean[nkeys*multiplier];
+ this.multiplier = multiplier;
}
public void aggregate(final ResultBitBuffer result, final Split split) {
- System.arraycopy(result.getResult(), 0, results, split.fromIndex,
- split.ntuples);
+ System.arraycopy(result.getResult(), 0, results,
+ split.fromIndex*multiplier,
+ split.ntuples*multiplier);
onCount.addAndGet(result.getOnCount());
Copied: branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/changesets/ChangeRecord.java (from rev 3729, branches/CHANGE_SET_BRANCH/bigdata-sails/src/java/com/bigdata/rdf/sail/changesets/ChangeRecord.java)
===================================================================
--- branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/changesets/ChangeRecord.java (rev 0)
+++ branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/changesets/ChangeRecord.java 2010-10-08 01:45:59 UTC (rev 3750)
@@ -0,0 +1,98 @@
+package com.bigdata.rdf.changesets;
+
+import java.util.Comparator;
+import com.bigdata.rdf.spo.ISPO;
+import com.bigdata.rdf.spo.SPOComparator;
+
+public class ChangeRecord implements IChangeRecord {
+
+ private final ISPO stmt;
+
+ private final ChangeAction action;
+
+// private final StatementEnum oldType;
+
+ public ChangeRecord(final ISPO stmt, final ChangeAction action) {
+
+// this(stmt, action, null);
+//
+// }
+//
+// public ChangeRecord(final BigdataStatement stmt, final ChangeAction action,
+// final StatementEnum oldType) {
+//
+ this.stmt = stmt;
+ this.action = action;
+// this.oldType = oldType;
+
+ }
+
+ public ChangeAction getChangeAction() {
+
+ return action;
+
+ }
+
+// public StatementEnum getOldStatementType() {
+//
+// return oldType;
+//
+// }
+
+ public ISPO getStatement() {
+
+ return stmt;
+
+ }
+
+ @Override
+ public boolean equals(Object o) {
+
+ if (o == this)
+ return true;
+
+ if (o == null || o instanceof IChangeRecord == false)
+ return false;
+
+ final IChangeRecord rec = (IChangeRecord) o;
+
+ final ISPO stmt2 = rec.getStatement();
+
+ // statements are equal
+ if (stmt == stmt2 ||
+ (stmt != null && stmt2 != null && stmt.equals(stmt2))) {
+
+ // actions are equal
+ return action == rec.getChangeAction();
+
+ }
+
+ return false;
+
+ }
+
+ public String toString() {
+
+ StringBuilder sb = new StringBuilder();
+
+ sb.append(action).append(": ").append(stmt);
+
+ return sb.toString();
+
+ }
+
+ public static final Comparator<IChangeRecord> COMPARATOR =
+ new Comparator<IChangeRecord>() {
+
+ public int compare(final IChangeRecord r1, final IChangeRecord r2) {
+
+ final ISPO spo1 = r1.getStatement();
+ final ISPO spo2 = r2.getStatement();
+
+ return SPOComparator.INSTANCE.compare(spo1, spo2);
+
+ }
+
+ };
+
+}
Copied: branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/changesets/IChangeLog.java (from rev 3668, branches/CHANGE_SET_BRANCH/bigdata-sails/src/java/com/bigdata/rdf/sail/changesets/IChangeLog.java)
===================================================================
--- branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/changesets/IChangeLog.java (rev 0)
+++ branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/changesets/IChangeLog.java 2010-10-08 01:45:59 UTC (rev 3750)
@@ -0,0 +1,38 @@
+package com.bigdata.rdf.changesets;
+
+/**
+ * Provides detailed information on changes made to statements in the database.
+ * Change records are generated for any statements that are used in
+ * addStatement() or removeStatements() operations on the SAIL connection, as
+ * well as any inferred statements that are added or removed as a result of
+ * truth maintenance when the database has inference enabled. Change records
+ * will be sent to an instance of this class via the
+ * {@link #changeEvent(IChangeRecord)} method. These events will
+ * occur on an ongoing basis as statements are added to or removed from the
+ * indices. It is the change log's responsibility to collect change records.
+ * When the transaction is actually committed (or aborted), the change log will
+ * receive notification via {@link #transactionCommited()} or
+ * {@link #transactionAborted()}.
+ */
+public interface IChangeLog {
+
+ /**
+ * Occurs when a statement add or remove is flushed to the indices (but
+ * not yet committed).
+ *
+ * @param record
+ * the {@link IChangeRecord}
+ */
+ void changeEvent(final IChangeRecord record);
+
+ /**
+ * Occurs when the current SAIL transaction is committed.
+ */
+ void transactionCommited();
+
+ /**
+ * Occurs if the current SAIL transaction is aborted.
+ */
+ void transactionAborted();
+
+}
Copied: branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/changesets/IChangeRecord.java (from rev 3729, branches/CHANGE_SET_BRANCH/bigdata-sails/src/java/com/bigdata/rdf/sail/changesets/IChangeRecord.java)
===================================================================
--- branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/changesets/IChangeRecord.java (rev 0)
+++ branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/changesets/IChangeRecord.java 2010-10-08 01:45:59 UTC (rev 3750)
@@ -0,0 +1,120 @@
+package com.bigdata.rdf.changesets;
+
+import com.bigdata.rdf.model.BigdataStatement;
+import com.bigdata.rdf.model.StatementEnum;
+import com.bigdata.rdf.spo.ISPO;
+
+/**
+ * Provides detailed information on changes made to statements in the database.
+ * Change records are generated for any statements that are used in
+ * addStatement() or removeStatements() operations on the SAIL connection, as
+ * well as any inferred statements that are added or removed as a result of
+ * truth maintenance when the database has inference enabled.
+ * <p>
+ * See {@link IChangeLog}.
+ */
+public interface IChangeRecord {
+
+ /**
+ * Attempting to add or remove statements can have a number of different
+ * effects. This enum captures the different actions that can take place as
+ * a result of trying to add or remove a statement from the database.
+ */
+ public enum ChangeAction {
+
+ /**
+ * The focus statement was not in the database before and will be
+ * in the database after the commit. This can be the result of either
+ * explicit addStatement() operations on the SAIL connection, or from
+ * new inferences being generated via truth maintenance when the
+ * database has inference enabled. If the focus statement has a
+ * statement type of explicit then it was added via an addStatement()
+ * operation. If the focus statement has a statement type of inferred
+ * then it was added via truth maintenance.
+ */
+ INSERTED,
+
+ /**
+ * The focus statement was in the database before and will not
+ * be in the database after the commit. When the database has inference
+ * and truth maintenance enabled, the statement that is the focus of
+ * this change record was either an explicit statement that was the
+ * subject of a removeStatements() operation on the connection, or it
+ * was an inferred statement that was removed as a result of truth
+ * maintenance. Either way, the statement is no longer provable as an
+ * inference using other statements still in the database after the
+ * commit. If it were still provable, the explicit statement would have
+ * had its type changed to inferred, and the inferred statement would
+ * have remained untouched by truth maintenance. If an inferred
+ * statement was the subject of a removeStatement() operation on the
+ * connection it would have resulted in a no-op, since inferences can
+ * only be removed via truth maintenance.
+ */
+ REMOVED,
+
+ /**
+ * This change action can only occur when inference and truth
+ * maintenance are enabled on the database. Sometimes an attempt at
+ * statement addition or removal via an addStatement() or
+ * removeStatements() operation on the connection will result in a type
+ * change rather than an actual assertion or deletion. When in
+ * inference mode, statements can have one of three statement types:
+ * explicit, inferred, or axiom (see {@link StatementEnum}). There are
+ * several reasons why a statement will change type rather than be
+ * asserted or deleted:
+ * <p>
+ * <ul>
+ * <li> A statement is asserted, but already exists in the database as
+ * an inference or an axiom. The existing statement will have its type
+ * changed from inference or axiom to explicit. </li>
+ * <li> An explicit statement is retracted, but is still provable by
+ * other means. It will have its type changed from explicit to
+ * inference. </li>
+ * <li> An explicit statement is retracted, but is one of the axioms
+ * needed for inference. It will have its type changed from explicit to
+ * axiom. </li>
+ * </ul>
+ */
+ UPDATED,
+
+// /**
+// * This change action can occur for one of two reasons:
+// * <p>
+// * <ul>
+// * <li> A statement is asserted, but already exists in the database as
+// * an explicit statement. </li>
+// * <li> An inferred statement or an axiom is retracted. Only explicit
+// * statements can be retracted via removeStatements() operations. </li>
+// * </ul>
+// */
+// NO_OP
+
+ }
+
+ /**
+ * Return the ISPO that is the focus of this change record.
+ *
+ * @return
+ * the {@link ISPO}
+ */
+ ISPO getStatement();
+
+ /**
+ * Return the change action for this change record.
+ *
+ * @return
+ * the {@link ChangeAction}
+ */
+ ChangeAction getChangeAction();
+
+// /**
+// * If the change action is {@link ChangeAction#TYPE_CHANGE}, this method
+// * will return the old statement type of the focus statement. The
+// * new statement type is available on the focus statement itself.
+// *
+// * @return
+// * the old statement type of the focus statement
+// */
+// StatementEnum getOldStatementType();
+
+}
Copied: branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/changesets/StatementWriter.java (from rev 3736, branches/CHANGE_SET_BRANCH/bigdata-sails/src/java/com/bigdata/rdf/sail/changesets/StatementWriter.java)
===================================================================
--- branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/changesets/StatementWriter.java (rev 0)
+++ branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/changesets/StatementWriter.java 2010-10-08 01:45:59 UTC (rev 3750)
@@ -0,0 +1,208 @@
+package com.bigdata.rdf.changesets;
+
+import java.util.Iterator;
+import java.util.Map;
+import org.apache.log4j.Logger;
+import com.bigdata.rdf.changesets.IChangeRecord.ChangeAction;
+import com.bigdata.rdf.internal.IV;
+import com.bigdata.rdf.model.BigdataBNode;
+import com.bigdata.rdf.spo.ISPO;
+import com.bigdata.rdf.spo.SPO;
+import com.bigdata.rdf.spo.ISPO.ModifiedEnum;
+import com.bigdata.rdf.store.AbstractTripleStore;
+import com.bigdata.relation.accesspath.IElementFilter;
+import com.bigdata.striterator.ChunkedArrayIterator;
+import com.bigdata.striterator.IChunkedOrderedIterator;
+
+public class StatementWriter {
+
+ protected static final Logger log = Logger.getLogger(StatementWriter.class);
+
+ public static long addStatements(final AbstractTripleStore database,
+ final AbstractTripleStore statementStore,
+ final boolean copyOnly,
+ final IElementFilter<ISPO> filter,
+ final IChunkedOrderedIterator<ISPO> itr,
+ final IChangeLog changeLog) {
+
+ long n = 0;
+
+ if (itr.hasNext()) {
+
+// final BigdataStatementIteratorImpl itr2 =
+// new BigdataStatementIteratorImpl(database, bnodes, itr)
+// .start(database.getExecutorService());
+//
+// final BigdataStatement[] stmts =
+// new BigdataStatement[database.getChunkCapacity()];
+ final SPO[] stmts = new SPO[database.getChunkCapacity()];
+
+ int i = 0;
+ while ((i = nextChunk(itr, stmts)) > 0) {
+ n += addStatements(database, statementStore, copyOnly, filter,
+ stmts, i, changeLog);
+ }
+
+ }
+
+ return n;
+
+ }
+
+ private static long addStatements(final AbstractTripleStore database,
+ final AbstractTripleStore statementStore,
+ final boolean copyOnly,
+ final IElementFilter<ISPO> filter,
+ final ISPO[] stmts,
+ final int numStmts,
+ final IChangeLog changeLog) {
+
+// final SPO[] tmp = allocateSPOs(stmts, numStmts);
+
+ final long n = database.addStatements(statementStore, copyOnly,
+ new ChunkedArrayIterator<ISPO>(numStmts, stmts,
+ null/* keyOrder */), filter);
+
+ // Copy the state of the isModified() flag and notify changeLog
+ for (int i = 0; i < numStmts; i++) {
+
+ if (stmts[i].isModified()) {
+
+// stmts[i].setModified(true);
+
+ if (changeLog != null) {
+
+ switch(stmts[i].getModified()) {
+ case INSERTED:
+ changeLog.changeEvent(new ChangeRecord(stmts[i], ChangeAction.INSERTED));
+ break;
+ case UPDATED:
+ changeLog.changeEvent(new ChangeRecord(stmts[i], ChangeAction.UPDATED));
+ break;
+ case REMOVED:
+ throw new AssertionError();
+ default:
+ break;
+ }
+
+ }
+
+ }
+
+ }
+
+ return n;
+
+ }
+
+ public static long removeStatements(final AbstractTripleStore database,
+ final IChunkedOrderedIterator<ISPO> itr,
+ final boolean computeClosureForStatementIdentifiers,
+ final IChangeLog changeLog) {
+
+ long n = 0;
+
+ if (itr.hasNext()) {
+
+// final BigdataStatementIteratorImpl itr2 =
+// new BigdataStatementIteratorImpl(database, bnodes, itr)
+// .start(database.getExecutorService());
+//
+// final BigdataStatement[] stmts =
+// new BigdataStatement[database.getChunkCapacity()];
+ final SPO[] stmts = new SPO[database.getChunkCapacity()];
+
+ int i = 0;
+ while ((i = nextChunk(itr, stmts)) > 0) {
+ n += removeStatements(database, stmts, i,
+ computeClosureForStatementIdentifiers, changeLog);
+ }
+
+ }
+
+ return n;
+
+ }
+
+ private static long removeStatements(final AbstractTripleStore database,
+ final ISPO[] stmts,
+ final int numStmts,
+ final boolean computeClosureForStatementIdentifiers,
+ final IChangeLog changeLog) {
+
+ final long n = database.removeStatements(
+ new ChunkedArrayIterator<ISPO>(numStmts, stmts,
+ null/* keyOrder */),
+ computeClosureForStatementIdentifiers);
+
+ // Copy the state of the isModified() flag and notify changeLog
+ for (int i = 0; i < numStmts; i++) {
+
+ if (stmts[i].isModified()) {
+
+ // just to be safe
+ stmts[i].setModified(ModifiedEnum.REMOVED);
+
+ changeLog.changeEvent(
+ new ChangeRecord(stmts[i], ChangeAction.REMOVED));
+
+ }
+
+ }
+
+ return n;
+
+ }
+
+ private static int nextChunk(final Iterator<ISPO> itr,
+ final ISPO[] stmts) {
+
+ assert stmts != null && stmts.length > 0;
+
+ int i = 0;
+ while (itr.hasNext()) {
+ stmts[i++] = itr.next();
+ if (i == stmts.length) {
+ // stmts[] is full
+ return i;
+ }
+ }
+
+ /*
+ * stmts[] is empty (i = 0) or partially
+ * full (i > 0 && i < stmts.length)
+ */
+ return i;
+
+ }
+
+// private static SPO[] allocateSPOs(final BigdataStatement[] stmts,
+// final int numStmts) {
+//
+// final SPO[] tmp = new SPO[numStmts];
+//
+// for (int i = 0; i < tmp.length; i++) {
+//
+// final BigdataStatement stmt = stmts[i];
+//
+// final SPO spo = new SPO(stmt);
+//
+// if (log.isDebugEnabled())
+// log.debug("writing: " + stmt.toString() + " (" + spo + ")");
+//
+// if(!spo.isFullyBound()) {
+//
+// throw new AssertionError("Not fully bound? : " + spo);
+//
+// }
+//
+// tmp[i] = spo;
+//
+// }
+//
+// return tmp;
+//
+//
+// }
+
+}
Modified: branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/inf/SPOAssertionBuffer.java
===================================================================
--- branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/inf/SPOAssertionBuffer.java 2010-10-07 20:23:05 UTC (rev 3749)
+++ branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/inf/SPOAssertionBuffer.java 2010-10-08 01:45:59 UTC (rev 3750)
@@ -35,9 +35,9 @@
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicLong;
+import com.bigdata.rdf.changesets.IChangeLog;
import com.bigdata.rdf.internal.IV;
import com.bigdata.rdf.model.BigdataBNode;
-import com.bigdata.rdf.sail.changesets.IChangeLog;
import com.bigdata.rdf.spo.ISPO;
import com.bigdata.rdf.spo.ISPOAssertionBuffer;
import com.bigdata.rdf.spo.JustificationWriter;
@@ -106,10 +106,11 @@
*/
protected final boolean justify;
+ /**
+ * Used for change set notification (optional).
+ */
protected final IChangeLog changeLog;
- protected final Map<IV, BigdataBNode> bnodes;
-
/**
* Create a buffer.
*
@@ -135,14 +136,35 @@
boolean justified) {
this(focusStore, db, filter, capacity, justified,
- null/* changeLog */, null/* bnodes */);
+ null/* changeLog */);
}
+ /**
+ * Create a buffer.
+ *
+ * @param focusStore
+ * The focusStore on which the entailments computed by closure
+ * will be written (required). This is either the database or a
+ * temporary focusStore used during incremental TM.
+ * @param db
+ * The database in which the terms are defined (required).
+ * @param filter
+ * Option filter. When present statements matched by the filter
+ * are NOT retained by the {@link SPOAssertionBuffer} and will
+ * NOT be added to the <i>focusStore</i>.
+ * @param capacity
+ * The maximum {@link SPO}s that the buffer can hold before it
+ * is {@link #flush()}ed.
+ * @param justified
+ * true iff the Truth Maintenance strategy requires that we
+ * focusStore {@link Justification}s for entailments.
+ * @param changeLog
+ * optional change log for change notification
+ */
public SPOAssertionBuffer(AbstractTripleStore focusStore,
AbstractTripleStore db, IElementFilter<ISPO> filter, int capacity,
- boolean justified,
- final IChangeLog changeLog, final Map<IV, BigdataBNode> bnodes
+ boolean justified, final IChangeLog changeLog
) {
super(db, filter, capacity);
@@ -163,8 +185,6 @@
this.changeLog = changeLog;
- this.bnodes = bnodes;
-
}
/**
@@ -214,15 +234,13 @@
} else {
- n = com.bigdata.rdf.sail.changesets.
- StatementWriter.addStatements(
+ n = com.bigdata.rdf.changesets.StatementWriter.addStatements(
db,
focusStore,
true/* copyOnly */,
null/* filter */,
new ChunkedArrayIterator<ISPO>(numStmts, stmts, null/*keyOrder*/),
- changeLog,
- bnodes);
+ changeLog);
}
@@ -249,7 +267,7 @@
tasks.add(new StatementWriter(getTermDatabase(), focusStore,
false/* copyOnly */, new ChunkedArrayIterator<ISPO>(
numStmts, stmts, null/*keyOrder*/), nwritten,
- changeLog, bnodes));
+ changeLog));
// task will write justifications on the justifications index.
final AtomicLong nwrittenj = new AtomicLong();
Modified: branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/inf/SPORetractionBuffer.java
===================================================================
--- branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/inf/SPORetractionBuffer.java 2010-10-07 20:23:05 UTC (rev 3749)
+++ branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/inf/SPORetractionBuffer.java 2010-10-08 01:45:59 UTC (rev 3750)
@@ -28,10 +28,10 @@
package com.bigdata.rdf.inf;
import java.util.Map;
+import com.bigdata.rdf.changesets.IChangeLog;
+import com.bigdata.rdf.changesets.StatementWriter;
import com.bigdata.rdf.internal.IV;
import com.bigdata.rdf.model.BigdataBNode;
-import com.bigdata.rdf.sail.changesets.IChangeLog;
-import com.bigdata.rdf.sail.changesets.StatementWriter;
import com.bigdata.rdf.spo.ISPO;
import com.bigdata.rdf.spo.SPO;
import com.bigdata.rdf.store.AbstractTripleStore;
@@ -54,11 +54,12 @@
private final AbstractTripleStore store;
private final boolean computeClosureForStatementIdentifiers;
-
+
+ /**
+ * Optional change log for change notification.
+ */
protected final IChangeLog changeLog;
- protected final Map<IV, BigdataBNode> bnodes;
-
/**
* @param store
* The database from which the statement will be removed when the
@@ -73,13 +74,25 @@
boolean computeClosureForStatementIdentifiers) {
this(store, capacity, computeClosureForStatementIdentifiers,
- null/* changeLog */, null/* bnodes */);
+ null/* changeLog */);
}
+ /**
+ * @param store
+ * The database from which the statement will be removed when the
+ * buffer is {@link #flush()}ed.
+ * @param capacity
+ * The capacity of the retraction buffer.
+ * @param computeClosureForStatementIdentifiers
+ * See
+ * {@link AbstractTripleStore#removeStatements(com.bigdata.rdf.spo.ISPOIterator, boolean)}
+ * @param changeLog
+ * optional change log for change notification
+ */
public SPORetractionBuffer(AbstractTripleStore store, int capacity,
boolean computeClosureForStatementIdentifiers,
- final IChangeLog changeLog, final Map<IV, BigdataBNode> bnodes) {
+ final IChangeLog changeLog) {
super(store, null/*filter*/, capacity);
@@ -92,8 +105,6 @@
this.changeLog = changeLog;
- this.bnodes = bnodes;
-
}
public int flush() {
@@ -114,8 +125,7 @@
new ChunkedArrayIterator<ISPO>(
numStmts,stmts,null/*keyOrder*/),
computeClosureForStatementIdentifiers,
- changeLog,
- bnodes);
+ changeLog);
}
Modified: branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/inf/TruthMaintenance.java
===================================================================
--- branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/inf/TruthMaintenance.java 2010-10-07 20:23:05 UTC (rev 3749)
+++ branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/inf/TruthMaintenance.java 2010-10-08 01:45:59 UTC (rev 3750)
@@ -54,13 +54,13 @@
import org.apache.log4j.MDC;
import com.bigdata.journal.TemporaryStore;
+import com.bigdata.rdf.changesets.IChangeLog;
import com.bigdata.rdf.internal.IV;
import com.bigdata.rdf.model.BigdataBNode;
import com.bigdata.rdf.model.BigdataStatement;
import com.bigdata.rdf.model.StatementEnum;
import com.bigdata.rdf.rio.IStatementBuffer;
import com.bigdata.rdf.rules.InferenceEngine;
-import com.bigdata.rdf.sail.changesets.IChangeLog;
import com.bigdata.rdf.spo.ExplicitSPOFilter;
import com.bigdata.rdf.spo.ISPO;
import com.bigdata.rdf.spo.SPO;
@@ -244,15 +244,39 @@
) {
return applyExistingStatements(focusStore, database, filter,
- null/* changeLog */, null/* bnodes */);
+ null/* changeLog */);
}
+
+ /**
+ * Any statements in the <i>fousStore</i> that are already in the database
+ * are converted to explicit statements (iff they are not already explicit)
+ * and <strong>removed</strong> from the <i>focusStore</i> as a
+ * side-effect. This prevents the application of the rules to data that is
+ * already known to the database.
+ *
+ * @param focusStore
+ * The store whose closure is being computed.
+ * @param database
+ * The database.
+ * @param filter
+ * An optional filter. Statements matching the filter are NOT
+ * written on the database, but they are still removed from the
+ * focusStore.
+ * @param changeLog
+ * optional change log for change notification
+ *
+ * @return The #of statements that were removed from the focusStore.
+ *
+ * @todo this uses some techniques that are not scaleable if the focusStore
+ * is extremely large.
+ */
static public int applyExistingStatements(
final AbstractTripleStore focusStore,
final AbstractTripleStore database,
final IElementFilter<ISPO> filter,
- final IChangeLog changeLog, final Map<IV, BigdataBNode> bnodes
+ final IChangeLog changeLog
) {
if(INFO)
@@ -286,7 +310,7 @@
final SPOAssertionBuffer assertionBuffer = new SPOAssertionBuffer(
database, database, filter, capacity, false/* justified */,
- changeLog, bnodes);
+ changeLog);
/*
* This buffer will retract statements from the tempStore that are
@@ -385,12 +409,26 @@
*/
public ClosureStats assertAll(final TempTripleStore tempStore) {
- return assertAll(tempStore, null/* changeLog */, null/* bnodes */);
+ return assertAll(tempStore, null/* changeLog */);
}
+ /**
+ * Perform truth maintenance for statement assertion.
+ * <p>
+ * This method computes the closure of the temporary store against the
+ * database, writing entailments into the temporary store. Once all
+ * entailments have been computed, it then copies the all statements in the
+ * temporary store into the database and deletes the temporary store.
+ *
+ * @param tempStore
+ * A temporary store containing statements to be asserted. The
+ * tempStore will be closed as a post-condition.
+ * @param changeLog
+ * optional change log for change notification
+ */
public ClosureStats assertAll(final TempTripleStore tempStore,
- final IChangeLog changeLog, final Map<IV, BigdataBNode> bnodes) {
+ final IChangeLog changeLog) {
if (tempStore == null) {
@@ -436,7 +474,8 @@
* consistent if we change our mind about that practice.
*/
- applyExistingStatements(tempStore, database, inferenceEngine.doNotAddFilter, changeLog, bnodes);
+ applyExistingStatements(tempStore, database,
+ inferenceEngine.doNotAddFilter, changeLog);
final ClosureStats stats = inferenceEngine.computeClosure(tempStore);
@@ -457,7 +496,7 @@
final long ncopied = tempStore.copyStatements(database,
null/* filter */, true /* copyJustifications */,
- changeLog, bnodes);
+ changeLog);
// database.dumpStore(database,true,true,false,true);
@@ -506,12 +545,35 @@
*/
public ClosureStats retractAll(final TempTripleStore tempStore) {
- return retractAll(tempStore, null/* changeLog */, null/* bnodes */);
+ return retractAll(tempStore, null/* changeLog */);
}
+ /**
+ * Perform truth maintenance for statement retraction.
+ * <p>
+ * When the closure is computed, each statement to be retracted is examined
+ * to determine whether or not it is still entailed by the database without
+ * the support of the statements that were explicitly retracted. Statements
+ * that were explicit in the database that are still provable are converted
+ * to inferences. Statements which can no longer be proven (i.e., that are
+ * not supported by a grounded {@link Justification} chain) are retracted
+ * from the database and added into another temporary store and their
+ * justifications are deleted from the database. This process repeats with
+ * the new temporary store until no fixed point (no more ungrounded
+ * statements are identified).
+ *
+ * @param tempStore
+ * A temporary store containing explicit statements to be
+ * retracted from the database. The tempStore will be closed and
+ * as a post-condition.
+ * @param changeLog
+ * optional change log for change notification
+ *
+ * @return statistics about the closure operation.
+ */
public ClosureStats retractAll(final TempTripleStore tempStore,
- final IChangeLog changeLog, final Map<IV, BigdataBNode> bnodes) {
+ final IChangeLog changeLog) {
final long begin = System.currentTimeMillis();
@@ -547,7 +609,7 @@
}
// do truth maintenance.
- retractAll(stats, tempStore, 0, changeLog, bnodes);
+ retractAll(stats, tempStore, 0, changeLog);
MDC.remove("depth");
@@ -624,10 +686,12 @@
* Recursive depth - this is ZERO(0) the first time the method is
* called. At depth ZERO(0) the tempStore MUST contain only the
* explicit statements to be retracted.
+ * @param changeLog
+ * optional change log for change notification
*/
private void retractAll(final ClosureStats stats,
final TempTripleStore tempStore, final int depth,
- final IChangeLog changeLog, final Map<IV, BigdataBNode> bnodes) {
+ final IChangeLog changeLog) {
MDC.put("depth", "depth=" + depth);
@@ -651,7 +715,8 @@
// final TempTripleStore focusStore = new TempTripleStore(database.getProperties(), database);
// consider each statement in the tempStore.
- final IChunkedOrderedIterator<ISPO> itr = tempStore.getAccessPath(SPOKeyOrder.SPO).iterator();
+ final IChunkedOrderedIterator<ISPO> itr =
+ tempStore.getAccessPath(SPOKeyOrder.SPO).iterator();
final long nretracted;
final long ndowngraded;
@@ -677,8 +742,7 @@
null, //filter @todo was inferenceEngine.doNotAddFilter,
capacity,//
false,// justify
- changeLog,
- bnodes
+ changeLog
);
/*
@@ -696,7 +760,7 @@
*/
final SPORetractionBuffer retractionBuffer = new SPORetractionBuffer(
database, capacity, false/* computeClosureForStatementIdentifiers */,
- changeLog, bnodes);
+ changeLog);
/*
* Note: when we enter this method recursively statements in the
@@ -1003,7 +1067,7 @@
* Recursive processing.
*/
- retractAll(stats, focusStore, depth + 1, changeLog, bnodes);
+ retractAll(stats, focusStore, depth + 1, changeLog);
}
Modified: branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/model/BigdataStatementImpl.java
===================================================================
--- branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/model/BigdataStatementImpl.java 2010-10-07 20:23:05 UTC (rev 3749)
+++ branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/model/BigdataStatementImpl.java 2010-10-08 01:45:59 UTC (rev 3750)
@@ -66,7 +66,8 @@
private StatementEnum type;
private boolean userFlag;
private transient boolean override = false;
- private transient boolean modified = false;
+// private transient boolean modified = false;
+ private transient ModifiedEnum modified = ModifiedEnum.NONE;
/**
* Used by {@link BigdataValueFactory}
@@ -227,7 +228,7 @@
return "<" + s + ", " + p + ", " + o + (c == null ? "" : ", " + c)
+ ">" + (type == null ? "" : " : " + type)
- + (modified ? " : modified" : "");
+ + (isModified() ? " : modified("+modified+")" : "");
}
@@ -353,14 +354,20 @@
public boolean isModified() {
- return modified;
+ return modified != ModifiedEnum.NONE;
}
- public void setModified(final boolean modified) {
+ public void setModified(final ModifiedEnum modified) {
this.modified = modified;
}
+
+ public ModifiedEnum getModified() {
+
+ return modified;
+
+ }
}
Modified: branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/rio/StatementBuffer.java
===================================================================
--- branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/rio/StatementBuffer.java 2010-10-07 20:23:05 UTC (rev 3749)
+++ branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/rio/StatementBuffer.java 2010-10-08 01:45:59 UTC (rev 3750)
@@ -40,6 +40,9 @@
import org.openrdf.model.URI;
import org.openrdf.model.Value;
+import com.bigdata.rdf.changesets.ChangeRecord;
+import com.bigdata.rdf.changesets.IChangeLog;
+import com.bigdata.rdf.changesets.IChangeRecord.ChangeAction;
import com.bigdata.rdf.internal.IV;
import com.bigdata.rdf.model.BigdataBNode;
import com.bigdata.rdf.model.BigdataResource;
@@ -48,9 +51,6 @@
import com.bigdata.rdf.model.BigdataValue;
import com.bigdata.rdf.model.BigdataValueFactory;
import com.bigdata.rdf.model.StatementEnum;
-import com.bigdata.rdf.sail.changesets.ChangeRecord;
-import com.bigdata.rdf.sail.changesets.IChangeLog;
-import com.bigdata.rdf.sail.changesets.IChangeRecord.ChangeAction;
import com.bigdata.rdf.spo.ISPO;
import com.bigdata.rdf.spo.SPO;
import com.bigdata.rdf.store.AbstractTripleStore;
@@ -879,28 +879,6 @@
// final long nwritten = writeSPOs(sids ? tmp.clone() : tmp, numStmts);
final long nwritten = writeSPOs(tmp.clone(), numStmts);
- // Copy the state of the isModified() flag
- {
-
- for (int i = 0; i < numStmts; i++) {
-
- if (tmp[i].isModified()) {
-
- stmts[i].setModified(true);
-
- if (changeLog != null) {
-
- changeLog.changeEvent(
- new ChangeRecord(stmts[i], ChangeAction.ADDED));
-
- }
-
- }
-
- }
-
- }
-
if (sids) {
/*
@@ -972,6 +950,34 @@
}
+ // Copy the state of the isModified() flag
+ for (int i = 0; i < numStmts; i++) {
+
+ if (tmp[i].isModified()) {
+
+ stmts[i].setModified(tmp[i].getModified());
+
+ if (changeLog != null) {
+
+ switch(stmts[i].getModified()) {
+ case INSERTED:
+ changeLog.changeEvent(new ChangeRecord(stmts[i], ChangeAction.INSERTED));
+ break;
+ case UPDATED:
+ changeLog.changeEvent(new ChangeRecord(stmts[i], ChangeAction.UPDATED));
+ break;
+ case REMOVED:
+ throw new AssertionError();
+ default:
+ break;
+ }
+
+ }
+
+ }
+
+ }
+
return nwritten;
}
Modified: branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/spo/ISPO.java
===================================================================
--- branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/spo/ISPO.java 2010-10-07 20:23:05 UTC (rev 3749)
+++ branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/spo/ISPO.java 2010-10-08 01:45:59 UTC (rev 3750)
@@ -240,7 +240,7 @@
* Modification can indicate that the statement was inserted, retracted, or
* had its associated {@link StatementEnum} in the database updated.
*/
- public void setModified(boolean modified);
+ public void setModified(ModifiedEnum modified);
/**
* Return the state of the transient <i>modified</i> flag. This flag
@@ -265,6 +265,8 @@
* Because this information is set at a low-level it can not currently
* be used in combination with truth maintenance mechanisms.
*/
+ public ModifiedEnum getModified();
+
public boolean isModified();
/**
@@ -294,5 +296,60 @@
* The database whose lexicon will be used.
*/
public String toString(IRawTripleStore db);
+
+ public enum ModifiedEnum {
+
+ INSERTED, REMOVED, UPDATED, NONE;
+
+ public static boolean[] toBooleans(final ModifiedEnum[] modified, final int n) {
+
+ final boolean[] b = new boolean[n*2];
+ for (int i = 0; i < n; i++) {
+ switch(modified[i]) {
+ case INSERTED:
+ b[i*2] = true;
+ b[i*2+1] = false;
+ break;
+ case REMOVED:
+ b[i*2] = false;
+ b[i*2+1] = true;
+ break;
+ case UPDATED:
+ b[i*2] = true;
+ b[i*2+1] = true;
+ break;
+ case NONE:
+ default:
+ b[i*2] = false;
+ b[i*2+1] = false;
+ break;
+ }
+ }
+
+ return b;
+
+ }
+
+ public static ModifiedEnum[] fromBooleans(final boolean[] b, final int n) {
+
+ assert n < b.length && n % 2 == 0;
+
+ final ModifiedEnum[] m = new ModifiedEnum[n/2];
+ for (int i = 0; i < n; i+=2) {
+ if (b[i] && !b[i+1])
+ m[i/2] = INSERTED;
+ else if (!b[i] && b[i+1])
+ m[i/2] = REMOVED;
+ else if (b[i] && b[i+1])
+ m[i/2] = UPDATED;
+ else
+ m[i/2] = NONE;
+ }
+
+ return m;
+
+ }
+
+ }
}
Modified: branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/spo/SPO.java
===================================================================
--- branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/spo/SPO.java 2010-10-07 20:23:05 UTC (rev 3749)
+++ branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/spo/SPO.java 2010-10-08 01:45:59 UTC (rev 3750)
@@ -88,7 +88,8 @@
*/
private transient boolean override = false;
- private transient boolean modified = false;
+// private transient boolean modified = false;
+ private transient ModifiedEnum modified = ModifiedEnum.NONE;
final public IV get(final int index) {
switch(index) {
@@ -621,7 +622,7 @@
+ (c == null ? "" : ", " + toString(c))
+ (type == null ? "" : " : " + type
+ (override ? ", override" : ""))
- + (modified ? ", modified" : "") + " >";
+ + (isModified() ? ", modified ("+modified+")" : "") + " >";
}
@@ -670,7 +671,7 @@
t = "Unknown ";
}
- return t + (modified ? "(*)" : "") + " : "
+ return t + (isModified() ? "(*)" : "") + " : "
+ store.toString(s, p, o, c);
} else {
@@ -713,14 +714,20 @@
public boolean isModified() {
- return modified;
+ return modified != ModifiedEnum.NONE;
}
- public void setModified(final boolean modified) {
+ public void setModified(final ModifiedEnum modified) {
this.modified = modified;
}
+
+ public ModifiedEnum getModified() {
+
+ return modified;
+
+ }
}
Modified: branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/spo/SPOIndexRemover.java
===================================================================
--- branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/spo/SPOIndexRemover.java 2010-10-07 20:23:05 UTC (rev 3749)
+++ branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/spo/SPOIndexRemover.java 2010-10-08 01:45:59 UTC (rev 3750)
@@ -12,6 +12,7 @@
import com.bigdata.btree.proc.AbstractKeyArrayIndexProcedure.ResultBitBufferHandler;
import com.bigdata.btree.proc.BatchRemove.BatchRemoveConstructor;
import com.bigdata.rdf.inf.Justification;
+import com.bigdata.rdf.spo.ISPO.ModifiedEnum;
/**
* Class writes on a statement index, removing the specified statements (batch
@@ -183,7 +184,7 @@
final boolean[] bits = modified.getResult();
writeCount = modified.getOnCount();
-
+
for (int i = 0; i < numStmts; i++) {
if (bits[i]) {
@@ -198,8 +199,8 @@
* explicitly cleared it in between those writes).
*/
- a[i].setModified(bits[i]);
-
+ a[i].setModified(ModifiedEnum.REMOVED);
+
}
}
Modified: branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/spo/SPOIndexWriteProc.java
===================================================================
--- branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/spo/SPOIndexWriteProc.java 2010-10-07 20:23:05 UTC (rev 3749)
+++ branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/spo/SPOIndexWriteProc.java 2010-10-08 01:45:59 UTC (rev 3750)
@@ -29,6 +29,7 @@
import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
+import java.util.Arrays;
import org.apache.log4j.Logger;
import com.bigdata.btree.BytesUtil;
import com.bigdata.btree.IIndex;
@@ -43,6 +44,7 @@
import com.bigdata.rdf.internal.TermId;
import com.bigdata.rdf.internal.VTE;
import com.bigdata.rdf.model.StatementEnum;
+import com.bigdata.rdf.spo.ISPO.ModifiedEnum;
import com.bigdata.rdf.store.IRawTripleStore;
import com.bigdata.relation.IMutableRelationIndexWriteProcedure;
@@ -190,7 +192,12 @@
// .endsWith(SPOKeyOrder.SPO.getIndexName()) : false;
// Array used to report by which statements were modified by this operation.
- final boolean[] modified = reportMutation ? new boolean[n] : null;
+ final ModifiedEnum[] modified = reportMutation ? new ModifiedEnum[n] : null;
+ if (reportMutation) {
+ for (int i = 0; i < n; i++) {
+ modified[i] = ModifiedEnum.NONE;
+ }
+ }
for (int i = 0; i < n; i++) {
@@ -255,7 +262,7 @@
writeCount++;
if (reportMutation)
- modified[i] = true;
+ modified[i] = ModifiedEnum.INSERTED;
} else {
@@ -291,7 +298,7 @@
writeCount++;
if (reportMutation)
- modified[i] = true;
+ modified[i] = ModifiedEnum.UPDATED;
}
@@ -337,7 +344,7 @@
writeCount++;
if (reportMutation)
- modified[i] = true;
+ modified[i] = ModifiedEnum.UPDATED;
}
@@ -351,9 +358,26 @@
log.info("Wrote " + writeCount + " SPOs on ndx="
+ ndx.getIndexMetadata().getName());
- return reportMutation ? new ResultBitBuffer(n, modified, writeCount)
- : Long.valueOf(writeCount);
-
+ if (reportMutation) {
+
+ final boolean[] b = ModifiedEnum.toBooleans(modified, n);
+
+ int onCount = 0;
+ for (int i = 0; i < b.length; i++) {
+ if (b[i])
+ onCount++;
+ }
+
+ ResultBitBuffer rbb = new ResultBitBuffer(b.length, b, onCount);
+
+ return rbb;
+
+ } else {
+
+ return Long.valueOf(writeCount);
+
+ }
+
}
/**
Modified: branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/spo/SPOIndexWriter.java
===================================================================
--- branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/spo/SPOIndexWriter.java 2010-10-07 20:23:05 UTC (rev 3749)
+++ branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/spo/SPOIndexWriter.java 2010-10-08 01:45:59 UTC (rev 3750)
@@ -57,6 +57,7 @@
import com.bigdata.btree.proc.AbstractKeyArrayIndexProcedure.ResultBitBuffer;
import com.bigdata.btree.proc.AbstractKeyArrayIndexProcedure.ResultBitBufferHandler;
import com.bigdata.io.ByteArrayBuffer;
+import com.bigdata.rdf.spo.ISPO.ModifiedEnum;
import com.bigdata.rdf.spo.SPOIndexWriteProc.IndexWriteProcConstructor;
import com.bigdata.relation.accesspath.IElementFilter;
@@ -283,7 +284,7 @@
*/
final long _begin = System.currentTimeMillis();
- final long writeCount;
+ long writeCount = 0;
if (reportMutation) {
/*
@@ -294,7 +295,7 @@
*/
final ResultBitBufferHandler aggregator = new ResultBitBufferHandler(
- numToAdd);
+ numToAdd,2);
ndx.submit(0/* fromIndex */, numToAdd/* toIndex */, keys, vals,
IndexWriteProcConstructor.REPORT_MUTATION, aggregator);
@@ -303,11 +304,11 @@
final boolean[] bits = modified.getResult();
- writeCount = modified.getOnCount();
-
+ final ModifiedEnum[] m = ModifiedEnum.fromBooleans(bits, bits.length);
+
for (int i = 0; i < numToAdd; i++) {
- if (bits[i]) {
+ if (m[i] != ModifiedEnum.NONE) {
/*
* Note: This only turns on the modified flag. It will not
@@ -319,7 +320,9 @@
* explicitly cleared it in between those writes).
*/
- denseStmts[i].setModified(bits[i]);
+ denseStmts[i].setModified(m[i]);
+
+ writeCount++;
}
Modified: branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/spo/StatementWriter.java
===================================================================
--- branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/spo/StatementWriter.java 2010-10-07 20:23:05 UTC (rev 3749)
+++ branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/spo/StatementWriter.java 2010-10-08 01:45:59 UTC (rev 3750)
@@ -4,12 +4,12 @@
import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.log4j.Logger;
+import com.bigdata.rdf.changesets.ChangeRecord;
+import com.bigdata.rdf.changesets.IChangeLog;
+import com.bigdata.rdf.changesets.IChangeRecord.ChangeAction;
import com.bigdata.rdf.internal.IV;
import com.bigdata.rdf.model.BigdataBNode;
import com.bigdata.rdf.model.BigdataStatement;
-import com.bigdata.rdf.sail.changesets.ChangeRecord;
-import com.bigdata.rdf.sail.changesets.IChangeLog;
-import com.bigdata.rdf.sail.changesets.IChangeRecord.ChangeAction;
import com.bigdata.rdf.store.AbstractTripleStore;
import com.bigdata.rdf.store.BigdataStatementIteratorImpl;
import com.bigdata.relation.accesspath.IElementFilter;
@@ -40,8 +40,6 @@
private final IChangeLog changeLog;
- private final Map<IV, BigdataBNode> bnodes;
-
/**
* @param database
* The database. If statement identifiers are being generated
@@ -67,14 +65,14 @@
IChunkedOrderedIterator<ISPO> itr, AtomicLong nwritten) {
this(database, statementStore, copyOnly, itr, nwritten,
- null/* changeLog */, null/* bnodes */);
+ null/* changeLog */);
}
public StatementWriter(final AbstractTripleStore database,
final AbstractTripleStore statementStore, final boolean copyOnly,
final IChunkedOrderedIterator<ISPO> itr, final AtomicLong nwritten,
- final IChangeLog changeLog, final Map<IV, BigdataBNode> bnodes) {
+ final IChangeLog changeLog) {
if (database == null)
throw new IllegalArgumentException();
@@ -100,8 +98,6 @@
this.changeLog = changeLog;
- this.bnodes = bnodes;
-
}
/**
@@ -122,14 +118,13 @@
} else {
- n = com.bigdata.rdf.sail.changesets.StatementWriter.addStatements(
+ n = com.bigdata.rdf.changesets.StatementWriter.addStatements(
database,
statementStore,
copyOnly,
null/* filter */,
itr,
- changeLog,
- bnodes);
+ changeLog);
}
Modified: branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/store/AbstractTripleStore.java
===================================================================
--- branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/store/AbstractTripleStore.java 2010-10-07 20:23:05 UTC (rev 3749)
+++ branches/CHANGE_SET_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/store/AbstractTripleStore.java 2010-10-08 01:45:59 UTC (rev 3750)
@@ -75,6 +75,7 @@
import com.bigdata.rdf.axioms.BaseAxioms;
import com.bigdata.rdf.axioms.NoAxioms;
import com.bigdata.rdf.axioms.OwlAxioms;
+import com.bigdata.rdf.changesets.IChangeLog;
import com.bigdata.rdf.inf.IJustificationIterator;
import com.bigdata.rdf.inf.Justification;
import com.bigdata.rdf.inf.JustificationIterator;
@@ -103,7 +104,6 @@
import com.bigdata.rdf.rules.MatchRule;
import com.bigdata.rdf.rules.RDFJoinNexusFactory;
import com.bigdata.rdf.rules.RuleContextEnum;
-import com.bigdata.rdf.sail.changesets.IChangeLog;
import com.bigdata.rdf.spo.BulkCompleteConverter;
import com.bigdata.rdf.spo.BulkFilterConverter;
import com.bigdata.rdf.spo.ExplicitSPOFilter;
@@ -2203,6 +2203,13 @@
}
+ final public BigdataStatement getStatement(final Statement s) {
+
+ return getStatement(s.getSubject(), s.getPredicate(),
+ s.getObject(), s.getContext());
+
+ }
+
final public BigdataStatement getStatement(final Resource s, final URI p,
final Value o) {
@@ -2987,7 +2994,7 @@
) {
return copyStatements(dst, filter, copyJustifications,
- null/* changeLog */, null /* bnodes */);
+ null/* changeLog */);
}
@@ -2995,7 +3002,7 @@
final AbstractTripleStore dst,//
final IElementFilter<ISPO> filter,//
final boolean copyJustifications,//
- final IChangeLog changeLog, final Map<IV, BigdataBNode> bnodes
+ final IChangeLog changeLog
) {
if (dst == this)
@@ -3017,15 +3024,13 @@
} else {
- return com.bigdata.rdf.sail.changesets.
- StatementWriter.addStatements(
+ return com.bigdata.rdf.changesets.StatementWriter.addStatements(
dst,
dst,
true/* copyOnly */,
null/* filter */,
itr,
- changeLog,
- bnodes);
+ changeLog);
}
@@ -3051,7 +3056,7 @@
// task will write SPOs on the statement indices.
tasks.add(new StatementWriter(dst, dst, true/* copyOnly */,
- itr, nwritten, changeLog, bnodes));
+ itr, nwritten, changeLog));
// task will write justifications on the justifications index.
final AtomicLong nwrittenj = new AtomicLong();
Modified: branches/CHANGE_SET_BRANCH/bigdata-sails/src/java/com/bigdata/rdf/sail/BigdataSail.java
===================================...
[truncated message content] |
