Smalyshev has uploaded a new change for review.
https://gerrit.wikimedia.org/r/237036
Change subject: Batch more queries together
......................................................................
Batch more queries together
Change-Id: Ic7a40e97c3ceb6ee00d7e076dcf96230a280865e
---
M tools/src/main/java/org/wikidata/query/rdf/tool/Update.java
M tools/src/main/java/org/wikidata/query/rdf/tool/change/Change.java
M
tools/src/main/java/org/wikidata/query/rdf/tool/rdf/RdfRepository.GetRevisions.sparql
M tools/src/main/java/org/wikidata/query/rdf/tool/rdf/RdfRepository.java
A
tools/src/main/java/org/wikidata/query/rdf/tool/rdf/RdfRepository.multiSync.sparql
M
tools/src/main/java/org/wikidata/query/rdf/tool/wikibase/WikibaseRepository.java
6 files changed, 194 insertions(+), 22 deletions(-)
git pull ssh://gerrit.wikimedia.org:29418/wikidata/query/rdf
refs/changes/36/237036/1
diff --git a/tools/src/main/java/org/wikidata/query/rdf/tool/Update.java
b/tools/src/main/java/org/wikidata/query/rdf/tool/Update.java
index 586d42e..ffba480 100644
--- a/tools/src/main/java/org/wikidata/query/rdf/tool/Update.java
+++ b/tools/src/main/java/org/wikidata/query/rdf/tool/Update.java
@@ -17,7 +17,6 @@
import java.util.Locale;
import java.util.Map;
import java.util.Set;
-import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
@@ -308,37 +307,37 @@
* changes
*/
private void handleChanges(Change.Batch batch) throws
InterruptedException, ExecutionException {
- List<Future<String>> tasks = new ArrayList<>();
+ List<Future<?>> tasks = new ArrayList<>();
Set<Change> trueChanges = getRevisionUpdates(batch);
long start = System.currentTimeMillis();
for (final Change change : trueChanges) {
- tasks.add(executor.submit(new Callable<String>() {
+ tasks.add(executor.submit(new Runnable() {
@Override
- public String call() {
+ public void run() {
while (true) {
try {
- return handleChange(change);
+ handleChange(change);
+ return;
} catch (RetryableException e) {
log.warn("Retryable error syncing. Retrying.", e);
} catch (ContainedException e) {
log.warn("Contained error syncing. Giving up on "
+ change.entityId(), e);
- return null;
+ return;
}
}
}
}));
}
- StringBuilder bigQuery = new StringBuilder();
- for (Future<String> task : tasks) {
- String query = task.get();
- if (query != null) {
- bigQuery.append(query);
- }
+
+ for (Future<?> task : tasks) {
+ task.get();
}
log.debug("Preparing update data took {} ms",
System.currentTimeMillis() - start);
- if (bigQuery.length() > 0) {
- rdfRepository.syncQuery(bigQuery.toString());
- }
+ rdfRepository.syncFromChanges(trueChanges);
+//
+// if (bigQuery.length() > 0) {
+// rdfRepository.syncQuery(bigQuery.toString());
+// }
updateMeter.mark(trueChanges.size());
}
@@ -374,10 +373,15 @@
}
log.debug("Filtered batch contains {} changes", trueChanges.size());
- repoValues = rdfRepository.getValues(changeIds);
- log.debug("Fetched {} values", repoValues.size());
- repoRefs = rdfRepository.getRefs(changeIds);
- log.debug("Fetched {} refs", repoRefs.size());
+ if (trueChanges.size() > 0) {
+ repoValues = rdfRepository.getValues(changeIds);
+ log.debug("Fetched {} values", repoValues.size());
+ repoRefs = rdfRepository.getRefs(changeIds);
+ log.debug("Fetched {} refs", repoRefs.size());
+ } else {
+ repoValues = null;
+ repoRefs = null;
+ }
return trueChanges;
}
@@ -418,7 +422,7 @@
* @throws RetryableException if there is a retryable error updating the
rdf
* store
*/
- private String handleChange(Change change) throws RetryableException {
+ private void handleChange(Change change) throws RetryableException {
log.debug("Processing data for {}", change);
Collection<Statement> statements =
wikibase.fetchRdfForEntity(change.entityId());
Set<String> values = new HashSet<>(repoValues.get(change.entityId()));
@@ -427,7 +431,9 @@
List<String> cleanupList = new ArrayList<>();
cleanupList.addAll(values);
cleanupList.addAll(refs);
- return rdfRepository.getSyncQuery(change.entityId(), statements,
cleanupList);
+ change.setStatements(statements);
+ change.setCleanupList(cleanupList);
+// return rdfRepository.getSyncQuery(change.entityId(), statements,
cleanupList);
}
/**
diff --git a/tools/src/main/java/org/wikidata/query/rdf/tool/change/Change.java
b/tools/src/main/java/org/wikidata/query/rdf/tool/change/Change.java
index 71473b0..239154c 100644
--- a/tools/src/main/java/org/wikidata/query/rdf/tool/change/Change.java
+++ b/tools/src/main/java/org/wikidata/query/rdf/tool/change/Change.java
@@ -2,9 +2,11 @@
import static com.google.common.base.Preconditions.checkNotNull;
+import java.util.Collection;
import java.util.Date;
import java.util.List;
+import org.openrdf.model.Statement;
import org.wikidata.query.rdf.tool.exception.RetryableException;
import com.google.common.collect.ImmutableList;
@@ -25,6 +27,16 @@
* Timestamp of the change.
*/
private final Date timestamp;
+
+ /**
+ * Set of processed statements for the change.
+ */
+ private Collection<Statement> statements;
+
+ /**
+ * Cleanup list for the change.
+ */
+ private Collection<String> cleanupList;
/**
* rcid of the change.
@@ -204,4 +216,35 @@
return (int)(rcid() - o.rcid());
}
+ /**
+ * Set statements collection.
+ * @return
+ */
+ public Collection<Statement> getStatements() {
+ return statements;
+ }
+
+ /**
+ * Return statements collection.
+ * @return
+ */
+ public void setStatements(Collection<Statement> statements) {
+ this.statements = statements;
+ }
+
+ /**
+ * Set cleanup list.
+ * @return
+ */
+ public Collection<String> getCleanupList() {
+ return cleanupList;
+ }
+
+ /**
+ * Return cleanup list.
+ * @param cleanupList
+ */
+ public void setCleanupList(Collection<String> cleanupList) {
+ this.cleanupList = cleanupList;
+ }
}
diff --git
a/tools/src/main/java/org/wikidata/query/rdf/tool/rdf/RdfRepository.GetRevisions.sparql
b/tools/src/main/java/org/wikidata/query/rdf/tool/rdf/RdfRepository.GetRevisions.sparql
index 8343246..e98206e 100644
---
a/tools/src/main/java/org/wikidata/query/rdf/tool/rdf/RdfRepository.GetRevisions.sparql
+++
b/tools/src/main/java/org/wikidata/query/rdf/tool/rdf/RdfRepository.GetRevisions.sparql
@@ -3,7 +3,7 @@
%values%
}
OPTIONAL {
- ?s %schema:about% ?repoRev
+ ?s %schema:version% ?repoRev
}
FILTER (!bound(?repoRev) || ?repoRev < ?rev)
}
diff --git
a/tools/src/main/java/org/wikidata/query/rdf/tool/rdf/RdfRepository.java
b/tools/src/main/java/org/wikidata/query/rdf/tool/rdf/RdfRepository.java
index f158298..1d84a53 100644
--- a/tools/src/main/java/org/wikidata/query/rdf/tool/rdf/RdfRepository.java
+++ b/tools/src/main/java/org/wikidata/query/rdf/tool/rdf/RdfRepository.java
@@ -94,6 +94,10 @@
*/
private final String syncBody;
/**
+ * Sparql for a portion of the update, batched sync.
+ */
+ private final String msyncBody;
+ /**
* Sparql for a portion of the update.
*/
private final String getValues;
@@ -135,6 +139,7 @@
public RdfRepository(URI uri, WikibaseUris uris) {
this.uri = uri;
this.uris = uris;
+ msyncBody = loadBody("multiSync");
syncBody = loadBody("sync");
updateLeftOffTimeBody = loadBody("updateLeftOffTime");
getValues = loadBody("GetValues");
@@ -321,6 +326,64 @@
}
/**
+ * Sync repository from changes list.
+ * @param changes List of changes.
+ * @return Number of triples modified.
+ */
+ public int syncFromChanges(Collection<Change> changes) {
+ if (changes.size() == 0) {
+ // no changes, we're done
+ return 0;
+ }
+ UpdateBuilder b = new UpdateBuilder(msyncBody);
+ b.bindUri("schema:about", SchemaDotOrg.ABOUT);
+ b.bindUri("prov:wasDerivedFrom", Provenance.WAS_DERIVED_FROM);
+ b.bind("uris.value", uris.value());
+ b.bind("uris.statement", uris.statement());
+ Set<String> entityIds = new HashSet<String>(changes.size());
+
+ List<Statement> insertStatements = new ArrayList<Statement>();
+ List<Statement> entityStatements = new ArrayList<Statement>();
+ Set<String> valueList = new HashSet<String>();
+
+ for (final Change change : changes) {
+ entityIds.add(change.entityId());
+ insertStatements.addAll(change.getStatements());
+
entityStatements.addAll(filtered(change.getStatements()).withSubject(uris.entity()
+ change.entityId()));
+ valueList.addAll(change.getCleanupList());
+ }
+
+ b.bindUris("entityList", entityIds, uris.entity());
+ b.bindStatements("insertStatements", insertStatements);
+ b.bindValues("entityStatements", entityStatements);
+
+ Collection<Statement> statementStatements =
filtered(insertStatements).withSubjectStarts(uris.statement());
+ b.bindValues("statementStatements", statementStatements);
+
+ Collection<Statement> aboutStatements = new
HashSet<Statement>(insertStatements);
+ aboutStatements.removeAll(entityStatements);
+ aboutStatements.removeAll(statementStatements);
+
aboutStatements.removeAll(filtered(insertStatements).withSubjectStarts(uris.value()));
+
aboutStatements.removeAll(filtered(insertStatements).withSubjectStarts(uris.reference()));
+ b.bindValues("aboutStatements", aboutStatements);
+
+ if (!valueList.isEmpty()) {
+ UpdateBuilder cleanup = new UpdateBuilder(cleanUnused);
+ cleanup.bindUris("values", valueList);
+ b.bind("cleanupQuery", cleanup.toString());
+ } else {
+ b.bind("cleanupQuery", "");
+ }
+
+ long start = System.currentTimeMillis();
+ int modified = execute("update", UPDATE_COUNT_RESPONSE, b.toString());
+ log.debug("Update query took {} millis and modified {} statements",
+ System.currentTimeMillis() - start, modified);
+ return modified;
+
+ }
+
+ /**
* Synchronizes the RDF repository's representation of an entity to be
* exactly the provided statements. You can think of the RDF managed for an
* entity as a tree rooted at the entity. The managed tree ends where the
diff --git
a/tools/src/main/java/org/wikidata/query/rdf/tool/rdf/RdfRepository.multiSync.sparql
b/tools/src/main/java/org/wikidata/query/rdf/tool/rdf/RdfRepository.multiSync.sparql
new file mode 100644
index 0000000..192a5d5
--- /dev/null
+++
b/tools/src/main/java/org/wikidata/query/rdf/tool/rdf/RdfRepository.multiSync.sparql
@@ -0,0 +1,55 @@
+# Clear out of date site links
+DELETE {
+ ?s ?p ?o .
+}
+WHERE {
+ VALUES ?entity {
+ %entityList%
+ }
+ ?s %schema:about% ?entity .
+ ?s ?p ?o .
+ # This construct is constantly reused throughout the updates. Its job is to
not delete statements
+ # that are still in use.
+ FILTER NOT EXISTS {
+ VALUES ( ?s ?p ?o ) {
+ %aboutStatements%
+ }
+ }
+};
+# Clear out of date statements about statements
+DELETE {
+ ?s ?p ?o .
+}
+WHERE {
+ VALUES ?entity {
+ %entityList%
+ }
+ ?entity ?statementPred ?s .
+ FILTER( STRSTARTS(STR(?s), "%uris.statement%") ) .
+ ?s ?p ?o .
+ FILTER NOT EXISTS {
+ VALUES ( ?s ?p ?o ) {
+ %statementStatements%
+ }
+ }
+};
+# Clear out of date statements about the entity
+DELETE {
+ ?entity ?p ?o .
+}
+WHERE {
+ VALUES ?entity {
+ %entityList%
+ }
+ ?entity ?p ?o .
+ FILTER NOT EXISTS {
+ VALUES ( ?entity ?p ?o ) {
+ %entityStatements%
+ }
+ }
+};
+# Insert new data
+INSERT {
+ %insertStatements%
+} WHERE {};
+%cleanupQuery%
diff --git
a/tools/src/main/java/org/wikidata/query/rdf/tool/wikibase/WikibaseRepository.java
b/tools/src/main/java/org/wikidata/query/rdf/tool/wikibase/WikibaseRepository.java
index 9cb7727..c507905 100644
---
a/tools/src/main/java/org/wikidata/query/rdf/tool/wikibase/WikibaseRepository.java
+++
b/tools/src/main/java/org/wikidata/query/rdf/tool/wikibase/WikibaseRepository.java
@@ -2,8 +2,10 @@
import java.io.IOException;
import java.io.InputStreamReader;
+import java.net.SocketException;
import java.net.URI;
import java.net.URISyntaxException;
+import java.net.UnknownHostException;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
@@ -121,6 +123,9 @@
}
parser.parse(new
InputStreamReader(response.getEntity().getContent(), Charsets.UTF_8),
uri.toString());
}
+ } catch (UnknownHostException | SocketException e) {
+ // We want to bail on this, since it happens to be sticky for some
reason
+ throw new RuntimeException(e);
} catch (IOException e) {
throw new RetryableException("Error fetching RDF for " + uri, e);
} catch (RDFParseException | RDFHandlerException e) {
--
To view, visit https://gerrit.wikimedia.org/r/237036
To unsubscribe, visit https://gerrit.wikimedia.org/r/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Ic7a40e97c3ceb6ee00d7e076dcf96230a280865e
Gerrit-PatchSet: 1
Gerrit-Project: wikidata/query/rdf
Gerrit-Branch: master
Gerrit-Owner: Smalyshev <[email protected]>
_______________________________________________
MediaWiki-commits mailing list
[email protected]
https://lists.wikimedia.org/mailman/listinfo/mediawiki-commits
