improvements to ingestion

Author: hmottestad

core/model/src/main/java/org/eclipse/rdf4j/model/impl/AbstractMemoryOverflowModel.java

@@ -243,9 +243,13 @@ public boolean add(Resource subj, IRI pred, Value obj, Resource... contexts) {
 		return add;
 	}
 
+	long count = 0;
+
 	@Override
 	public boolean add(Statement st) {
-		checkMemoryOverflow();
+		if (count++ % BATCH_SIZE == 0) {
+			checkMemoryOverflow();
+		}
 		boolean add = getDelegate().add(st);
 		if (add && memory instanceof DynamicModel) {
 			((DynamicModel) memory).maybeRegisterLargeStatementSetForReuse((DynamicModel) memory);

core/sail/base/src/main/java/org/eclipse/rdf4j/sail/base/Changeset.java

@@ -31,6 +31,7 @@
 
 import org.eclipse.rdf4j.common.annotation.Experimental;
 import org.eclipse.rdf4j.common.annotation.InternalUseOnly;
+import org.eclipse.rdf4j.common.transaction.IsolationLevel;
 import org.eclipse.rdf4j.common.transaction.IsolationLevels;
 import org.eclipse.rdf4j.model.IRI;
 import org.eclipse.rdf4j.model.Model;
@@ -122,6 +123,7 @@ public abstract class Changeset implements SailSink, ModelFactory {
 	private volatile boolean statementCleared;
 
 	private boolean closed;
+	private IsolationLevel sinkIsolationLevel = IsolationLevels.NONE;
 
 	public static boolean isOrderIndependent(Changeset changeset1, Changeset changeset2) {
 		Objects.requireNonNull(changeset1, "changeset1");
@@ -539,6 +541,7 @@ protected void setChangeset(Changeset from) {
 		assert !closed;
 		assert !from.closed;
 
+		this.sinkIsolationLevel = from.sinkIsolationLevel;
 		this.observed = from.observed;
 		this.approved = from.approved;
 		this.approvedEmpty = from.approvedEmpty;
@@ -552,6 +555,14 @@ protected void setChangeset(Changeset from) {
 		this.statementCleared = from.statementCleared;
 	}
 
+	IsolationLevel getSinkIsolationLevel() {
+		return sinkIsolationLevel;
+	}
+
+	void setSinkIsolationLevel(IsolationLevel sinkIsolationLevel) {
+		this.sinkIsolationLevel = Objects.requireNonNull(sinkIsolationLevel);
+	}
+
 	/**
 	 * Create a shallow clone of this Changeset. The shallow clone does not clone the underlying data structures, this
 	 * means that any changes made to the original will potentially be reflected in the clone and vice versa.

core/sail/base/src/main/java/org/eclipse/rdf4j/sail/base/SailSourceBranch.java

@@ -269,6 +269,7 @@ public Model createEmptyModel() {
 				return modelFactory.createEmptyModel();
 			}
 		};
+		changeset.setSinkIsolationLevel(level);
 		try {
 			semaphore.lock();
 			pending.add(changeset);
@@ -315,7 +316,7 @@ public void prepare() throws SailException {
 			semaphore.lock();
 			if (!changes.isEmpty()) {
 				if (prepared == null && serializable == null) {
-					prepared = backingSource.sink(IsolationLevels.NONE);
+					prepared = backingSource.sink(backingSinkIsolationLevel());
 				} else if (prepared == null) {
 					prepared = serializable;
 				}
@@ -327,6 +328,16 @@ public void prepare() throws SailException {
 		}
 	}
 
+	private IsolationLevel backingSinkIsolationLevel() {
+		for (Changeset change : changes) {
+			IsolationLevel level = change.getSinkIsolationLevel();
+			if (!IsolationLevels.NONE.isCompatibleWith(level)) {
+				return level;
+			}
+		}
+		return IsolationLevels.NONE;
+	}
+
 	@Override
 	public void flush() throws SailException {
 		try {