You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@jena.apache.org by an...@apache.org on 2025/06/09 10:38:31 UTC
(jena) 02/02: Spatial indexer: restricted graphs, list visible graphs, separated tasks from servlet, empty dataset, warn if srs or index file not set, disabled geometry serde, consolidated buildOrLoad methods, cleaned logging.
This is an automated email from the ASF dual-hosted git repository.
andy pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/jena.git
commit 2523f5a7ac64a747af050e4d805de16c55497a48
Author: Claus Stadler <Ra...@googlemail.com>
AuthorDate: Sun May 18 11:58:41 2025 +0200
Spatial indexer: restricted graphs, list visible graphs, separated tasks from servlet, empty dataset, warn if srs or index file not set, disabled geometry serde, consolidated buildOrLoad methods, cleaned logging.
---
.../java/org/apache/jena/sparql/util/Context.java | 5 +-
.../spatial/index/SpatialIndexCurrent.java | 6 +-
.../org/apache/jena/fuseki/build/FusekiConfig.java | 2 +-
.../fuseki/mod/geosparql/FMod_SpatialIndexer.java | 56 +-
.../mod/geosparql/SpatialIndexComputeService.java | 583 -------------------
.../mod/geosparql/SpatialIndexerService.java | 625 +++++++++++++++++++++
.../jena/fuseki/mod/geosparql/TaskTracker.java | 128 -----
.../src/main/resources/spatial-indexer/index.html | 35 +-
.../mod/geosparql/TestFMod_SpatialIndexer.java | 11 +-
.../mod/geosparql/TestSpatialIndexerTasks.java | 209 +++++++
.../MainPlaygroundFusekiModGeoSparql.java | 46 --
.../jena/geosparql/assembler/GeoAssembler.java | 25 +-
.../geosparql/configuration/GeoSPARQLConfig.java | 14 +-
.../configuration/GeoSPARQLOperations.java | 60 +-
.../geo/topological/GenericPropertyFunction.java | 8 +-
.../geosparql/spatial/SpatialIndexConstants.java} | 18 +-
.../geosparql/spatial/SpatialIndexFindUtils.java | 50 +-
.../geosparql/spatial/index/v1/SpatialIndexV1.java | 9 +-
.../index/v2/KryoRegistratorSpatialIndexV2.java | 14 +-
.../spatial/index/v2/STRtreePerGraph.java | 7 +-
.../index/v2/STRtreePerGraphSerializer.java | 4 +-
.../geosparql/spatial/index/v2/STRtreeUtils.java | 21 +-
.../spatial/index/v2/SpatialIndexIoKryo.java | 157 ++++--
.../spatial/index/v2/SpatialIndexLib.java | 409 ++++++++++++++
.../spatial/index/v2/SpatialIndexUtils.java | 215 -------
.../index/v2/SpatialIndexerComputation.java | 66 ++-
.../GenericSpatialPropertyFunction.java | 6 +-
.../geosparql/spatial/task/AbortableThread.java | 199 -------
.../task/{TaskControl.java => BasicTask.java} | 44 +-
.../task/TaskControlOverAbortableThread.java | 105 ----
.../jena/geosparql/spatial/task/TaskState.java} | 18 +-
.../jena/geosparql/spatial/task/TaskThread.java | 222 ++++++++
.../geosparql/geo/topological/CancelQueryTest.java | 6 +-
.../topological/GenericPropertyFunctionTest.java | 6 +-
.../geosparql/spatial/SpatialIndexTestData.java | 4 +-
.../spatial/TestSpatialIndexGraphLookupV2.java | 4 +-
.../spatial/index/v2/SpatialIndexTest.java | 2 +-
.../GenericSpatialPropertyFunctionTest.java | 4 +-
38 files changed, 1857 insertions(+), 1546 deletions(-)
diff --git a/jena-arq/src/main/java/org/apache/jena/sparql/util/Context.java b/jena-arq/src/main/java/org/apache/jena/sparql/util/Context.java
index 7a410e5b40..fd2a5f261e 100644
--- a/jena-arq/src/main/java/org/apache/jena/sparql/util/Context.java
+++ b/jena-arq/src/main/java/org/apache/jena/sparql/util/Context.java
@@ -390,8 +390,9 @@ public class Context {
}
/** Atomic compute. */
- public <V> Object compute(Symbol key, BiFunction<Symbol, Object, V> remappingFunction) {
- return context.compute(key, remappingFunction);
+ @SuppressWarnings("unchecked")
+ public <V> V compute(Symbol key, BiFunction<Symbol, Object, ? extends V> remappingFunction) {
+ return (V)context.compute(key, remappingFunction);
}
@Override
diff --git a/jena-benchmarks/jena-benchmarks-jmh/src/test/java/org/apache/jena/geosparql/spatial/index/SpatialIndexCurrent.java b/jena-benchmarks/jena-benchmarks-jmh/src/test/java/org/apache/jena/geosparql/spatial/index/SpatialIndexCurrent.java
index 3817c467d3..0a9f25afd0 100644
--- a/jena-benchmarks/jena-benchmarks-jmh/src/test/java/org/apache/jena/geosparql/spatial/index/SpatialIndexCurrent.java
+++ b/jena-benchmarks/jena-benchmarks-jmh/src/test/java/org/apache/jena/geosparql/spatial/index/SpatialIndexCurrent.java
@@ -24,7 +24,7 @@ import org.apache.jena.geosparql.configuration.GeoSPARQLOperations;
import org.apache.jena.geosparql.spatial.SpatialIndexException;
import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexIoKryo;
import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexPerGraph;
-import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexUtils;
+import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexLib;
import org.apache.jena.query.DatasetFactory;
import org.apache.jena.riot.Lang;
import org.apache.jena.riot.RDFParserBuilder;
@@ -60,7 +60,7 @@ public class SpatialIndexCurrent
@Override
public void init() {
- SpatialIndexUtils.setSpatialIndex(dsg.getContext(), null);
+ SpatialIndexLib.setSpatialIndex(dsg.getContext(), null);
}
@Override
@@ -73,7 +73,7 @@ public class SpatialIndexCurrent
@Override
public void build() throws Exception {
indexFile = Files.createTempFile("jena-", ".spatial-index");
- indexA = (SpatialIndexPerGraph)SpatialIndexUtils.buildSpatialIndex(dsg, finalSrs);
+ indexA = (SpatialIndexPerGraph)SpatialIndexLib.buildSpatialIndex(dsg, finalSrs);
SpatialIndexIoKryo.save(indexFile, indexA);
}
diff --git a/jena-fuseki2/jena-fuseki-core/src/main/java/org/apache/jena/fuseki/build/FusekiConfig.java b/jena-fuseki2/jena-fuseki-core/src/main/java/org/apache/jena/fuseki/build/FusekiConfig.java
index aff8113213..63eb901bf5 100644
--- a/jena-fuseki2/jena-fuseki-core/src/main/java/org/apache/jena/fuseki/build/FusekiConfig.java
+++ b/jena-fuseki2/jena-fuseki-core/src/main/java/org/apache/jena/fuseki/build/FusekiConfig.java
@@ -433,7 +433,7 @@ public class FusekiConfig {
return dataAccess;
} catch (FusekiException ex) {
Fuseki.configLog.error("Skipping: Failed to build service for "+BuildLib.displayStr(configuration, n));
- Fuseki.configLog.error(" "+ex.getMessage());
+ Fuseki.configLog.error(" "+ex.getMessage(), ex);
return null;
}
}
diff --git a/jena-fuseki2/jena-fuseki-mod-geosparql/src/main/java/org/apache/jena/fuseki/mod/geosparql/FMod_SpatialIndexer.java b/jena-fuseki2/jena-fuseki-mod-geosparql/src/main/java/org/apache/jena/fuseki/mod/geosparql/FMod_SpatialIndexer.java
index d80e9c41d9..b17f50925a 100644
--- a/jena-fuseki2/jena-fuseki-mod-geosparql/src/main/java/org/apache/jena/fuseki/mod/geosparql/FMod_SpatialIndexer.java
+++ b/jena-fuseki2/jena-fuseki-mod-geosparql/src/main/java/org/apache/jena/fuseki/mod/geosparql/FMod_SpatialIndexer.java
@@ -35,21 +35,10 @@ import org.apache.jena.rdf.model.Model;
public class FMod_SpatialIndexer implements FusekiAutoModule {
- private Operation spatialOperation = null;
-
- public Operation getOperation() {
- if (spatialOperation == null) {
- synchronized (this) {
- if (spatialOperation == null) {
- Fuseki.configLog.info(name() + ": Add spatial indexer operation into global registry.");
- spatialOperation = Operation.alloc("http://org.apache.jena/spatial-index-service",
- "spatial-indexer",
- "Spatial index computation service");
- }
- }
- }
- return spatialOperation;
- }
+ public static Operation spatialIndexerOperation =
+ Operation.alloc("http://jena.apache.org/fuseki#spatial-indexer",
+ "spatial-indexer",
+ "Spatial indexer service");
public FMod_SpatialIndexer() {
super();
@@ -66,9 +55,8 @@ public class FMod_SpatialIndexer implements FusekiAutoModule {
@Override
public void prepare(FusekiServer.Builder builder, Set<String> datasetNames, Model configModel) {
- Fuseki.configLog.info(name() + ": Registering spatial index operation and servlet.");
- Operation op = getOperation();
- builder.registerOperation(op, new SpatialIndexComputeService());
+ Fuseki.configLog.info(name() + ": Registering operation " + spatialIndexerOperation.getId());
+ builder.registerOperation(spatialIndexerOperation, new SpatialIndexerService());
}
/**
@@ -79,6 +67,21 @@ public class FMod_SpatialIndexer implements FusekiAutoModule {
public void configured(FusekiServer.Builder serverBuilder, DataAccessPointRegistry dapRegistry, Model configModel) {
FusekiAutoModule.super.configured(serverBuilder, dapRegistry, configModel);
+ boolean autoConfigure = false;
+ if (autoConfigure) {
+ autoConfigure(serverBuilder, dapRegistry, configModel);
+ }
+ }
+
+ /**
+ * Disabled for now.
+ *
+ * Automatically creates a corresponding spatial indexer endpoint for each SPARQL update endpoint.
+ * The spatial indexer endpoint with name follows the pattern '{updateEndpointName}-spatial and inherits the update endpoint's auth policy.
+ */
+ private void autoConfigure(FusekiServer.Builder serverBuilder, DataAccessPointRegistry dapRegistry, Model configModel) {
+ FusekiAutoModule.super.configured(serverBuilder, dapRegistry, configModel);
+
List<DataAccessPoint> newDataAccessPoints = new ArrayList<>();
// Register the spatial indexer for each update endpoint and inherit its auth policy.
for (DataAccessPoint dap : dapRegistry.accessPoints()) {
@@ -92,19 +95,18 @@ public class FMod_SpatialIndexer implements FusekiAutoModule {
List<Endpoint> spatialEndpoints = new ArrayList<>();
for (Endpoint updateEndpoint : updateEndpoints) {
String updateEndpointName = updateEndpoint.getName();
- String spatialEndpointName = updateEndpointName + "-spatial";
+ String geoIndexerEndpointName = updateEndpointName + "-spatial";
AuthPolicy authPolicy = updateEndpoint.getAuthPolicy();
- Fuseki.configLog.info(logPrefix + "Registering spatial indexer endpoint: " + spatialEndpointName);
+ Fuseki.configLog.info(logPrefix + "Registering spatial indexer endpoint: " + geoIndexerEndpointName);
- Operation op = getOperation();
- Endpoint spatialEndpoint = Endpoint.create()
- .operation(op)
- .endpointName(spatialEndpointName)
+ Endpoint geoIndexerEndpoint = Endpoint.create()
+ .operation(spatialIndexerOperation)
+ .endpointName(geoIndexerEndpointName)
.authPolicy(authPolicy)
.build();
- spatialEndpoints.add(spatialEndpoint);
+ spatialEndpoints.add(geoIndexerEndpoint);
}
// Create new DataService based on existing one with the spatial indexer endpoints attached.
@@ -120,8 +122,4 @@ public class FMod_SpatialIndexer implements FusekiAutoModule {
dapRegistry.register(dap);
});
}
-
- @Override
- public void serverAfterStarting(FusekiServer server) {
- }
}
diff --git a/jena-fuseki2/jena-fuseki-mod-geosparql/src/main/java/org/apache/jena/fuseki/mod/geosparql/SpatialIndexComputeService.java b/jena-fuseki2/jena-fuseki-mod-geosparql/src/main/java/org/apache/jena/fuseki/mod/geosparql/SpatialIndexComputeService.java
deleted file mode 100644
index 1f96e5ba8f..0000000000
--- a/jena-fuseki2/jena-fuseki-mod-geosparql/src/main/java/org/apache/jena/fuseki/mod/geosparql/SpatialIndexComputeService.java
+++ /dev/null
@@ -1,583 +0,0 @@
-/*
- * 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.jena.fuseki.mod.geosparql;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.io.PrintWriter;
-import java.nio.charset.StandardCharsets;
-import java.nio.file.Path;
-import java.time.Instant;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Iterator;
-import java.util.LinkedHashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentHashMap.KeySetView;
-import java.util.stream.Stream;
-
-import org.apache.commons.io.IOUtils;
-import org.apache.commons.lang3.exception.ExceptionUtils;
-import org.apache.jena.atlas.iterator.Iter;
-import org.apache.jena.atlas.lib.DateTimeUtils;
-import org.apache.jena.fuseki.FusekiException;
-import org.apache.jena.fuseki.servlets.BaseActionREST;
-import org.apache.jena.fuseki.servlets.HttpAction;
-import org.apache.jena.geosparql.spatial.SpatialIndex;
-import org.apache.jena.geosparql.spatial.SpatialIndexException;
-import org.apache.jena.geosparql.spatial.index.v2.STRtreePerGraph;
-import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexIoKryo;
-import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexPerGraph;
-import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexUtils;
-import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexerComputation;
-import org.apache.jena.geosparql.spatial.task.AbortableThread;
-import org.apache.jena.geosparql.spatial.task.TaskControl;
-import org.apache.jena.geosparql.spatial.task.TaskControlOverAbortableThread;
-import org.apache.jena.graph.Node;
-import org.apache.jena.graph.NodeFactory;
-import org.apache.jena.riot.WebContent;
-import org.apache.jena.riot.web.HttpNames;
-import org.apache.jena.sparql.core.DatasetGraph;
-import org.apache.jena.sparql.core.Quad;
-import org.apache.jena.sparql.util.Context;
-import org.apache.jena.system.Txn;
-import org.apache.jena.web.HttpSC;
-import org.locationtech.jts.index.strtree.STRtree;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.gson.Gson;
-import com.google.gson.JsonElement;
-import com.google.gson.JsonObject;
-import com.google.gson.reflect.TypeToken;
-
-import jakarta.servlet.AsyncContext;
-import jakarta.servlet.http.HttpServletRequest;
-import jakarta.servlet.http.HttpServletResponse;
-
-/**
- * Spatial index (re)computation service.
- *
- * Supports two types of tasks that are executed concurrently. For a given data set, only a single
- * task can be active at a given time. The status of the most recent task can be queried via the
- * rest API. Task execution also broadcasts start/abort/termination events via server side events
- * (SSE).
- *
- * <ul>
- * <li>Updating/replacing the graphs of a spatial index.</li>
- * <li>Removal of graphs from a spatial index that are absent in the corresponding data set.</li>
- * </ul>
- */
-public class SpatialIndexComputeService extends BaseActionREST {
- private static final Logger logger = LoggerFactory.getLogger(SpatialIndexComputeService.class);
-
- private static Gson gson = new Gson();
-
- /** Registered clients listening to server side events for indexer status updates. */
- private KeySetView<AsyncContext, ?> eventListeners = ConcurrentHashMap.newKeySet();
-
- public SpatialIndexComputeService() {}
-
- private static List<Node> extractGraphs(DatasetGraph dsg, HttpAction action) {
- String uris = action.getRequest().getParameter(HttpNames.paramGraph);
- Collection<String> strs;
- if (uris == null || uris.isBlank()) {
- strs = List.of(Quad.defaultGraphIRI.toString(), Quad.unionGraph.toString());
- } else {
- TypeToken<List<String>> typeToken = new TypeToken<List<String>>(){};
- strs = gson.fromJson(uris, typeToken);
- }
- List<Node> rawGraphNodes = strs.stream().map(NodeFactory::createURI).distinct().toList();
- // If the set of specified graphs is empty then index all.
- if (rawGraphNodes.isEmpty()) {
- rawGraphNodes = List.of(Quad.defaultGraphIRI, Quad.unionGraph);
- }
- List<Node> expandedGraphNodes = rawGraphNodes.stream()
- .flatMap(node -> expandUnionGraphNode(dsg, node).stream()).distinct().toList();
- return expandedGraphNodes;
- }
-
- private static boolean isReplaceMode(HttpAction action) {
- String str = action.getRequest().getParameter("replaceMode");
- boolean result = (str == null || str.isBlank()) ? false : Boolean.parseBoolean(str);
- return result;
- }
-
- private static int getThreadCount(HttpAction action) {
- String str = action.getRequest().getParameter("maxThreadCount");
- int result = (str == null || str.isBlank()) ? 1 : Integer.parseInt(str);
-
- if (result == 0) {
- result = Runtime.getRuntime().availableProcessors();
- }
-
- return result;
- }
-
- private static List<Node> expandUnionGraphNode(DatasetGraph dsg, Node node) {
- return Quad.isUnionGraph(node)
- ? listGraphNodes(dsg)
- : List.of(node);
- }
-
- private static List<Node> listGraphNodes(DatasetGraph dsg) {
- try (Stream<Node> s = Iter.asStream(dsg.listGraphNodes())) {
- return s.toList();
- }
- }
-
- /**
- * The GET command can serve: the website, the notification stream from task execution
- * and the latest task execution status.
- */
- @Override
- protected void doGet(HttpAction action) {
- String rawCommand = action.getRequestParameter("command");
- String command = Optional.ofNullable(rawCommand).orElse("website");
- switch (command) {
- case "website": serveWebSite(action); break;
- case "events": serveEvents(action); break;
- case "status": serveStatus(action); break;
- default:
- throw new UnsupportedOperationException("Unsupported operation: " + command);
- }
- }
-
- protected void serveWebSite(HttpAction action) {
- // Serves the minimal graphql ui
- String resourceName = "spatial-indexer/index.html";
- String str = null;
- try (InputStream in = SpatialIndexComputeService.class.getClassLoader().getResourceAsStream(resourceName)) {
- str = IOUtils.toString(in, StandardCharsets.UTF_8);
- } catch (IOException e) {
- throw new FusekiException(e);
- }
-
- if (str == null) {
- action.setResponseStatus(HttpSC.INTERNAL_SERVER_ERROR_500);
- action.setResponseContentType(WebContent.contentTypeTextPlain);
- str = "Failed to load classpath resource " + resourceName;
- } else {
- action.setResponseStatus(HttpSC.OK_200);
- action.setResponseContentType(WebContent.contentTypeHTML);
- }
- try (OutputStream out = action.getResponseOutputStream()) {
- IOUtils.write(str, out, StandardCharsets.UTF_8);
- } catch (IOException e) {
- throw new FusekiException(e);
- }
- }
-
- protected TaskControl<?> getActiveTask(HttpAction action) {
- DatasetGraph dsg = action.getDataset();
- Context cxt = dsg.getContext();
- TaskControl<?> activeTask = cxt.get(SpatialIndexUtils.SPATIAL_INDEX_TASK_SYMBOL);
- return activeTask;
- }
-
- /**
- * Post request: Handle API call.
- * Request is rejected if there is an already running task.
- */
- @Override
- protected void doPost(HttpAction action) {
- try {
- String rawCommand = action.getRequestParameter("command");
- String command = Optional.ofNullable(rawCommand).orElse("none");
- switch (command) {
- case "index": doIndex(action); break;
- case "clean": doClean(action); break;
- case "status": serveStatus(action); break;
- case "cancel": doStop(action); break;
- default: throw new UnsupportedOperationException("Unsupported operation: " + command);
- }
- } catch (Throwable t) {
- String str = ExceptionUtils.getStackTrace(t);
- action.log.error("An unexpected error occurred.", t);
- // FusekiException
- setResponse(action, HttpSC.INTERNAL_SERVER_ERROR_500, WebContent.contentTypeTextPlain, str);
- }
- }
-
- protected void serveEvents(HttpAction action) {
- HttpServletRequest request = action.getRequest();
- HttpServletResponse response = action.getResponse();
-
- response.setContentType("text/event-stream");
- response.setCharacterEncoding("UTF-8");
- response.setHeader("Cache-Control", "no-cache");
-
- final AsyncContext asyncContext = request.startAsync();
- asyncContext.setTimeout(0);
- eventListeners.add(asyncContext);
- }
-
- /**
- * Remove all graphs from the index for which there is no corresponding graph in the dataset.
- */
- protected void doClean(HttpAction action) throws Exception {
- DatasetGraph dsg = action.getDataset();
- Context cxt = dsg.getContext();
-
- TaskControlOverAbortableThread<Thread> taskCtl = new TaskControlOverAbortableThread<>("Spatial Indexer Task");
-
- cxt.compute(SpatialIndexUtils.SPATIAL_INDEX_TASK_SYMBOL, (key, priorTaskObj) -> {
- TaskControl<?> priorTask = (TaskControl<?>)priorTaskObj;
- if (priorTask != null && !priorTask.isComplete()) {
- throw new RuntimeException("A spatial indexing task is already active for this dataset. Wait for completion or abort it.");
- }
-
- AbortableThread<?> thread = new AbortableThread<>() {
- @Override
- public void runActual() throws Exception {
- broadcastTaskStart(getStartTime(), null);
-
- SpatialIndex spatialIndexRaw = SpatialIndexUtils.getSpatialIndex(cxt);
- if (spatialIndexRaw == null) {
- throw new SpatialIndexException("No spatial index available on current dataset.");
- } else if (spatialIndexRaw instanceof SpatialIndexPerGraph spatialIndex) {
- STRtreePerGraph perGraphIndex = spatialIndex.getIndex();
- Map<Node, STRtree> treeMap = perGraphIndex.getTreeMap();
- Set<Node> graphNodes = new LinkedHashSet<>(Txn.calculateRead(dsg, () -> listGraphNodes(dsg)));
- // int initialGraphCount = treeMap.keySet().size();
- List<Node> indexGraphNodes = new ArrayList<>(treeMap.keySet());
-
- int cleanCount = 0;
- for (Node node : indexGraphNodes) {
- if (!(node == null || Quad.isDefaultGraph(node)) && !graphNodes.contains(node)) {
- perGraphIndex.removeTree(node);
- ++cleanCount;
- // System.out.println("Removed: [" + node + "] " + (node == null));
- }
- }
-
- int finalGraphCount = treeMap.keySet().size();
- Path targetFile = spatialIndex.getLocation();
- // SpatialIndexUtils.setSpatialIndex(cxt, newIndex);
- if (cleanCount > 0 && targetFile != null) {
- // newIndex.setLocation(targetFile);
- action.log.info("Writing spatial index of {} graphs (cleaned: {}) to disk at path {}", finalGraphCount, cleanCount, targetFile.toAbsolutePath());
- SpatialIndexIoKryo.save(targetFile, spatialIndex);
- }
- String statusMsg = String.format("Updated spatial index of %d graphs (cleaned: %d)", finalGraphCount, cleanCount);
- setStatusMessage(statusMsg);
- if (logger.isInfoEnabled()) {
- logger.info("Indexing of {} graphs completed successfully.", finalGraphCount);
- }
- } else {
- throw new SpatialIndexException("Unsupported spatial index type for cleaning.");
- }
- }
-
- @Override
- protected void doAfterRun() {
- broadcastTaskEnd(getEndTime(), getThrowable(), getStatusMessage());
- }
- };
-
- taskCtl.setThread(thread);
- taskCtl.setSource(thread);
- thread.start();
- return taskCtl;
- });
- }
-
- protected void doStop(HttpAction action) {
- TaskControl<?> task = getActiveTask(action);
- String state;
- if (task != null) {
- state = "true";
- // Add a completion listener to return when done.
- task.abort();
- Thread thread = (Thread)task.getSource();
- try {
- thread.join();
- } catch (InterruptedException e) {
- // TODO Raise some HTTP error
- throw new RuntimeException(e);
- }
- } else {
- state = "false";
- }
-
- action.setResponseStatus(HttpSC.OK_200);
- action.setResponseContentType(WebContent.contentTypeJSON);
- try {
- action.getResponseOutputStream().print(String.format("{ \"stopped\": %s }", state));
- } catch (IOException e) {
- throw new FusekiException(e);
- }
- }
-
- /**
- * Serves a JSON object that captures the current indexing status.
- *
- * <pre>
- * {
- * isIndexing: Boolean, // Whether an index task is running.
- * isAborting: Boolean, // Iff isIndexing==true: Whether abort has been requested.
- * error: String, // Iff isIndexing==false: An error message of the most recently terminated indexing task.
- * message: String // Status message of the most recent indexing task (running or terminated).
- * time: long // EpochMillis of the last status update.
- * }
- * </pre>
- */
- protected void serveStatus(HttpAction action) {
- TaskControl<?> task = getActiveTask(action);
-
- JsonObject status = new JsonObject();
- Instant time;
- if (task == null) {
- status.addProperty("isIndexing", false);
- time = Instant.ofEpochMilli(0);
- } else {
- status.addProperty("isIndexing", !task.isComplete());
- if (!task.isComplete()) {
- status.addProperty("isAborting", task.isAborting());
- time = !task.isAborting() ? task.getStartTime() : task.getCancelTime();
- } else {
- time = task.getEndTime();
- }
- Throwable throwable = task.getThrowable();
- if (throwable != null) {
- String msg = ExceptionUtils.getStackTrace(throwable);
- status.addProperty("error", msg);
- }
-
- String msg = task.getStatusMessage();
- if (msg != null) {
- status.addProperty("message", msg);
- }
- }
- status.addProperty("time", time.toEpochMilli());
-
- action.setResponseStatus(HttpSC.OK_200);
- action.setResponseContentType(WebContent.contentTypeJSON);
- try {
- String str = gson.toJson(status);
- action.getResponseOutputStream().println(str);
- } catch (IOException e) {
- throw new FusekiException(e);
- }
- }
-
- protected TaskControl<?> scheduleTask(HttpAction action, SpatialIndexerComputation indexComputation, Path targetFile, boolean isReplaceTask) {
- DatasetGraph dsg = action.getDataset();
- Context cxt = dsg.getContext();
-
- TaskControlOverAbortableThread<Thread> taskCtl = new TaskControlOverAbortableThread<>("Spatial Indexer Task");
-
- cxt.compute(SpatialIndexUtils.SPATIAL_INDEX_TASK_SYMBOL, (key, priorTaskObj) -> {
- TaskControl<?> priorTask = (TaskControl<?>)priorTaskObj;
- if (priorTask != null && !priorTask.isComplete()) {
- throw new RuntimeException("A spatial indexing task is already active for this dataset. Wait for completion or abort it.");
- }
-
- long graphCount = indexComputation.getGraphNodes().size();
-
- AbortableThread<?> thread = new AbortableThread<Object>() {
- @Override
- public void runActual() throws Exception {
- broadcastTaskStart(getStartTime(), null);
- if (logger.isInfoEnabled()) {
- String replaceMsg = isReplaceTask ? "The resulting index will REPLACE a prior index." : "A prior index will be UPDATED with the newly indexed graphs.";
- logger.info("Indexing of {} graphs started. " + replaceMsg, graphCount);
- }
-
- // Uncomment to test artificial delays.
- // Thread.sleep(5000);
-
- SpatialIndexPerGraph newIndex = indexComputation.call();
-
- // If NOT in replace mode, add all graph-indexes from the previous index
- if (!isReplaceTask) {
- SpatialIndexPerGraph oldIndex = (SpatialIndexPerGraph)SpatialIndexUtils.getSpatialIndex(cxt);
- Map<Node, STRtree> treeMap = oldIndex.getIndex().getTreeMap();
- treeMap.forEach((name, tree) -> {
- if (!newIndex.getIndex().contains(name)) {
- newIndex.getIndex().setTree(name, tree);
- }
- });
- }
-
- SpatialIndexUtils.setSpatialIndex(cxt, newIndex);
- if (targetFile != null) {
- newIndex.setLocation(targetFile);
- action.log.info("Writing spatial index of {} graphs to disk at path {}", graphCount, targetFile.toAbsolutePath());
- SpatialIndexIoKryo.save(targetFile, newIndex);
- }
- String statusMsg = String.format("Updated spatial index with %d graphs.", graphCount);
- setStatusMessage(statusMsg);
- if (logger.isInfoEnabled()) {
- logger.info("Indexing of {} graphs completed successfully.", graphCount);
- }
- }
-
- @Override
- public void requestCancel() {
- broadcastTaskAbort(getCancelTime(), null);
- indexComputation.abort();
- super.requestCancel(); // Interrupt
- }
-
- @Override
- protected void doAfterRun() {
- try {
- indexComputation.close();
- } finally {
- try {
- Throwable throwable = getThrowable();
- Instant endTime = getEndTime();
- broadcastTaskEnd(endTime, throwable, getStatusMessage());
- } finally {
- if (logger.isInfoEnabled()) {
- logger.info("Indexing task of {} graphs terminated.", graphCount);
- }
- }
- }
- }
- };
-
- taskCtl.setThread(thread);
- taskCtl.setSource(thread);
- thread.start();
- return taskCtl;
- });
-
- return taskCtl;
- }
-
- protected void doIndex(HttpAction action) throws Exception {
- DatasetGraph dsg = action.getDataset();
- SpatialIndex index = SpatialIndexUtils.getSpatialIndex(dsg.getContext());
-
- if (index == null) { // error: no spatial index has been configured
- // XXX Could still allow for creating an in-memory-only index.
- String msg = String.format("[%d] no spatial index has been configured for the dataset", action.id);
- action.log.error(msg);
- action.setResponseStatus(HttpSC.SERVICE_UNAVAILABLE_503);
- action.setResponseContentType(WebContent.contentTypeTextPlain);
- try {
- action.getResponseWriter().println(msg);
- } catch (IOException e) {
- throw new FusekiException(e);
- }
- return;
- } else {
- boolean isReplaceMode = isReplaceMode(action);
- int threadCount = getThreadCount(action);
-
- // Only SpatialIndexPerGraph can be updated.
- // Check if the index can be updated. If not then raise an exception
- // that informs that only replace mode can be used in this situation.
- if (!isReplaceMode && !(index instanceof SpatialIndexPerGraph)) {
- throw new RuntimeException("Cannot update existing spatial index because its type is unsupported. Consider replacing the index.");
- }
-
- Path oldLocation = index.getLocation();
- if (oldLocation == null) {
- action.log.warn("Spatial index will not be persisted because no file location was configured.");
- }
-
- String srsURI = index.getSrsInfo().getSrsURI();
-
- List<Node> graphNodes = Txn.calculateRead(dsg, () -> extractGraphs(dsg, action));
- SpatialIndexerComputation task = new SpatialIndexerComputation(dsg, srsURI, graphNodes, threadCount);
-
- action.log.info(String.format("[%d] spatial index: computation request accepted.", action.id));
- scheduleTask(action, task, oldLocation, isReplaceMode);
-
- setResponse(action, HttpSC.OK_200, WebContent.contentTypeTextPlain, "Spatial index computation task accepted at " + DateTimeUtils.nowAsXSDDateTimeString());
- }
- }
-
- protected void setResponse(HttpAction action, int statusCode, String contentType, String payload) {
- action.setResponseStatus(statusCode);
- action.setResponseContentType(contentType);
- try (OutputStream out = action.getResponseOutputStream()) {
- IOUtils.write(payload, out, StandardCharsets.UTF_8);
- } catch (IOException e) {
- throw new FusekiException(e);
- }
- }
-
- protected void broadcastTaskStart(Instant time, String msg) {
- JsonObject json = new JsonObject();
- json.addProperty("isIndexing", true);
- json.addProperty("time", time.toEpochMilli());
- if (msg != null) {
- json.addProperty("message", msg);
- }
- broadcastJson(json);
- }
- protected void broadcastTaskAbort(Instant time, String msg) {
- JsonObject json = new JsonObject();
- json.addProperty("isIndexing", true);
- json.addProperty("isAborting", true);
- json.addProperty("time", time.toEpochMilli());
- if (msg != null) {
- json.addProperty("message", msg);
- }
- broadcastJson(json);
- }
-
- protected void broadcastTaskEnd(Instant time, Throwable throwable, String msg) {
- JsonObject json = new JsonObject();
- json.addProperty("isIndexing", false);
- json.addProperty("time", time.toEpochMilli());
- if (msg != null) {
- json.addProperty("message", msg);
- }
- if (throwable != null) {
- json.addProperty("error", ExceptionUtils.getStackTrace(throwable));
- }
- broadcastJson(json);
- }
-
- protected void broadcastJson(JsonElement jsonData) {
- String str = gson.toJson(jsonData);
- broadcastLine(str);
- }
-
- /**
- * Broadcast a payload to all registered listeners.
- * @param payload A string without newline characters.
- */
- protected void broadcastLine(String payload) {
- Iterator<AsyncContext> it = eventListeners.getMap().keySet().iterator();
- while (it.hasNext()) {
- AsyncContext context = it.next();
- try {
- PrintWriter writer = context.getResponse().getWriter();
- // Format demanded by server side events is: "data: <payload>\n\n".
- writer.println("data: " + payload);
- writer.println();
- writer.flush();
- } catch (IOException e) {
- it.remove();
- context.complete();
- }
- }
- }
-}
diff --git a/jena-fuseki2/jena-fuseki-mod-geosparql/src/main/java/org/apache/jena/fuseki/mod/geosparql/SpatialIndexerService.java b/jena-fuseki2/jena-fuseki-mod-geosparql/src/main/java/org/apache/jena/fuseki/mod/geosparql/SpatialIndexerService.java
new file mode 100644
index 0000000000..87eaa33aa1
--- /dev/null
+++ b/jena-fuseki2/jena-fuseki-mod-geosparql/src/main/java/org/apache/jena/fuseki/mod/geosparql/SpatialIndexerService.java
@@ -0,0 +1,625 @@
+/*
+ * 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.jena.fuseki.mod.geosparql;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.OutputStreamWriter;
+import java.io.PrintWriter;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.IdentityHashMap;
+import java.util.Iterator;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.jena.atlas.io.IO;
+import org.apache.jena.atlas.lib.DateTimeUtils;
+import org.apache.jena.fuseki.server.Endpoint;
+import org.apache.jena.fuseki.servlets.BaseActionREST;
+import org.apache.jena.fuseki.servlets.HttpAction;
+import org.apache.jena.fuseki.servlets.ServletOps;
+import org.apache.jena.geosparql.spatial.SpatialIndex;
+import org.apache.jena.geosparql.spatial.SpatialIndexConstants;
+import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexLib;
+import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexPerGraph;
+import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexerComputation;
+import org.apache.jena.geosparql.spatial.task.BasicTask;
+import org.apache.jena.geosparql.spatial.task.BasicTask.TaskListener;
+import org.apache.jena.graph.Node;
+import org.apache.jena.graph.NodeFactory;
+import org.apache.jena.query.Query;
+import org.apache.jena.query.QueryFactory;
+import org.apache.jena.riot.WebContent;
+import org.apache.jena.riot.web.HttpNames;
+import org.apache.jena.sparql.core.DatasetGraph;
+import org.apache.jena.sparql.core.Quad;
+import org.apache.jena.sparql.core.Var;
+import org.apache.jena.sparql.engine.binding.Binding;
+import org.apache.jena.sparql.exec.QueryExec;
+import org.apache.jena.sparql.exec.RowSet;
+import org.apache.jena.sparql.syntax.syntaxtransform.QueryTransformOps;
+import org.apache.jena.sparql.util.Context;
+import org.apache.jena.system.Txn;
+import org.apache.jena.web.HttpSC;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.gson.Gson;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import com.google.gson.reflect.TypeToken;
+import com.google.gson.stream.JsonWriter;
+
+import jakarta.servlet.AsyncContext;
+import jakarta.servlet.AsyncEvent;
+import jakarta.servlet.AsyncListener;
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletResponse;
+
+/**
+ * Spatial index (re)computation service.
+ *
+ * Supports two types of tasks that are executed concurrently. For a given data set, only a single
+ * task can be active at a given time. The status of the most recent task can be queried via the
+ * rest API. Task execution also broadcasts start/abort/termination events via server side events
+ * (SSE).
+ *
+ * <ul>
+ * <li>Updating/replacing the graphs of a spatial index.</li>
+ * <li>Removal of graphs from a spatial index that are absent in the corresponding data set.</li>
+ * </ul>
+ */
+public class SpatialIndexerService extends BaseActionREST {
+ private static final Logger logger = LoggerFactory.getLogger(SpatialIndexerService.class);
+
+ /** Help class for SSE listeners per endpoint. */
+ private class EndpointClients {
+ Map<AsyncContext, Runnable> eventListeners = Collections.synchronizedMap(new IdentityHashMap<>());
+ }
+
+ /** Gson configured to not emit newlines (pretty printing = off). */
+ private static Gson gsonForSse = new Gson();
+
+ /** Registered clients listening to server side events for indexer status updates. */
+ private Map<Endpoint, EndpointClients> listenersByEndpoint = Collections.synchronizedMap(new IdentityHashMap<>());
+
+
+ /** Constants for serving graph listings via sparql. */
+ private static final Var graphVar = Var.alloc("g");
+ private static final Var keywordVar = Var.alloc("keyword");
+ private static final Query allGraphsQuery = QueryFactory.create("SELECT ?g { GRAPH ?g { } } ORDER BY ASC(?g)");
+ private static final Query graphsByKeywordQuery = QueryFactory.create("SELECT ?g { GRAPH ?g { } FILTER(contains(lcase(str(?g)), ?keyword)) } ORDER BY ASC(?g)");
+
+ public SpatialIndexerService() {}
+
+ private static <T, C extends Collection<Node>> Set<Node> extractGraphsFromRequest(DatasetGraph dsg, HttpAction action) {
+ String uris = action.getRequest().getParameter(HttpNames.paramGraph);
+ Collection<String> strs;
+ if (uris == null || uris.isBlank()) {
+ strs = List.of(Quad.defaultGraphIRI.toString(), Quad.unionGraph.toString());
+ } else {
+ TypeToken<List<String>> typeToken = new TypeToken<List<String>>(){};
+ strs = gsonForSse.fromJson(uris, typeToken);
+ }
+ List<Node> rawGraphNodes = strs.stream().map(NodeFactory::createURI).distinct().toList();
+ // If the set of specified graphs is empty then index all.
+ if (rawGraphNodes.isEmpty()) {
+ rawGraphNodes = List.of(Quad.defaultGraphIRI, Quad.unionGraph);
+ }
+
+ Set<Node> uniqueGraphNodes = rawGraphNodes.stream()
+ .flatMap(node -> expandUnionGraphNode(new ArrayList<>(), dsg, node).stream())
+ .collect(Collectors.toCollection(LinkedHashSet::new));
+ return uniqueGraphNodes;
+ }
+
+ private static boolean isReplaceMode(HttpAction action) {
+ String str = action.getRequest().getParameter("replaceMode");
+ boolean result = (str == null || str.isBlank()) ? false : Boolean.parseBoolean(str);
+ return result;
+ }
+
+ private static int getThreadCount(HttpAction action) {
+ String str = action.getRequest().getParameter("maxThreadCount");
+ int result = (str == null || str.isBlank()) ? 1 : Integer.parseInt(str);
+
+ if (result == 0) {
+ result = Runtime.getRuntime().availableProcessors();
+ }
+
+ return result;
+ }
+
+ private static <C extends Collection<Node>> C expandUnionGraphNode(C accGraphs, DatasetGraph dsg, Node node) {
+ if (Quad.isUnionGraph(node)) {
+ SpatialIndexLib.accGraphNodes(accGraphs, dsg);
+ } else {
+ accGraphs.add(node);
+ }
+ return accGraphs;
+ }
+
+ /**
+ * The GET command can serve: the website, the notification stream from task execution
+ * and the latest task execution status.
+ */
+ @Override
+ protected void doGet(HttpAction action) {
+ String rawCommand = action.getRequestParameter("command");
+ String command = Optional.ofNullable(rawCommand).orElse("webpage");
+ switch (command) {
+ case "webpage": serveWebPage(action); break;
+ case "events": serveEvents(action); break;
+ case "status": serveStatus(action); break;
+ default:
+ throw new UnsupportedOperationException("Unsupported command (via HTTP GET): " + command);
+ }
+ }
+
+ protected void serveWebPage(HttpAction action) {
+ // Serves the minimal graphql ui
+ String resourceName = "spatial-indexer/index.html";
+ String str = null;
+ try (InputStream in = SpatialIndexerService.class.getClassLoader().getResourceAsStream(resourceName)) {
+ str = IOUtils.toString(in, StandardCharsets.UTF_8);
+ } catch (IOException e) {
+ ServletOps.errorOccurred(e.getMessage(), e);
+ }
+
+ if (str == null) {
+ ServletOps.error(HttpSC.INTERNAL_SERVER_ERROR_500, "Failed to load classpath resource " + resourceName);
+ } else {
+ action.setResponseStatus(HttpSC.OK_200);
+ action.setResponseContentType(WebContent.contentTypeHTML);
+ try (OutputStream out = action.getResponseOutputStream()) {
+ IOUtils.write(str, out, StandardCharsets.UTF_8);
+ } catch (IOException e) {
+ ServletOps.errorOccurred(e);
+ }
+ }
+ }
+
+ protected BasicTask getActiveTask(HttpAction action) {
+ DatasetGraph dsg = action.getDataset();
+ Context cxt = dsg.getContext();
+ BasicTask activeTask = cxt.get(SpatialIndexConstants.symSpatialIndexTask);
+ return activeTask;
+ }
+
+ /**
+ * Post request: Handle API call.
+ * Request is rejected if there is an already running task.
+ */
+ @Override
+ protected void doPost(HttpAction action) {
+ try {
+ String rawCommand = action.getRequestParameter("command");
+ String command = Optional.ofNullable(rawCommand).orElse("none");
+ switch (command) {
+ case "index": doIndex(action); break;
+ case "clean": doClean(action); break;
+ case "status": serveStatus(action); break;
+ case "cancel": doCancel(action); break;
+ case "graphs": serveGraphs(action); break;
+ default:
+ throw new UnsupportedOperationException("Unsupported command (via HTTP POST): " + command);
+ }
+ } catch (Throwable t) {
+ action.log.error("An unexpected error occurred.", t);
+ ServletOps.errorOccurred(t);
+ }
+ }
+
+ /**
+ * Serves a JSON array of the graph IRIs visible to the authenticated user.
+ * Non-IRI graph names are omitted in the output.
+ *
+ * Note: It is intentional to NOT expose a full SPARQL endpoint for security reasons.
+ */
+ protected void serveGraphs(HttpAction action) throws IOException {
+ long offset = Optional.ofNullable(action.getRequestParameter("offset")).map(Long::parseLong).orElse(Query.NOLIMIT);
+ long limit = Optional.ofNullable(action.getRequestParameter("limit")).map(Long::parseLong).orElse(Query.NOLIMIT);
+ String keyword = action.getRequestParameter("keyword");
+
+ Query query;
+ if (keyword != null) {
+ query = graphsByKeywordQuery.cloneQuery();
+ query = QueryTransformOps.syntaxSubstitute(query, Map.of(keywordVar, NodeFactory.createLiteralString(keyword)));
+ } else {
+ query = allGraphsQuery.cloneQuery();
+ }
+
+ if (offset != Query.NOLIMIT || limit != Query.NOLIMIT) {
+ query.setLimit(limit);
+ query.setOffset(offset);
+ }
+
+ action.beginRead();
+ try {
+ DatasetGraph dsg = action.getActiveDSG();
+ // XXX Make tracked should the exec tracker PR become available.
+ try (QueryExec qe = QueryExec.dataset(dsg).query(query).build()) {
+ try (JsonWriter jsonWriter = gsonForSse.newJsonWriter(
+ new OutputStreamWriter(action.getResponseOutputStream(), StandardCharsets.UTF_8))) {
+ RowSet rs = qe.select();
+ jsonWriter.beginArray();
+ jsonWriter.value(Quad.defaultGraphIRI.getURI());
+ jsonWriter.value(Quad.unionGraph.getURI());
+ while (rs.hasNext()) {
+ Binding b = rs.next();
+ Node n = b.get(graphVar);
+ if (n.isURI()) {
+ String uri = n.getURI();
+ jsonWriter.value(uri);
+ }
+ }
+ jsonWriter.endArray();
+ jsonWriter.flush();
+ }
+ }
+ } finally {
+ action.endRead();
+ }
+ }
+
+ protected void serveEvents(HttpAction action) {
+ HttpServletRequest request = action.getRequest();
+ HttpServletResponse response = action.getResponse();
+
+ response.setContentType("text/event-stream");
+ response.setCharacterEncoding("UTF-8");
+ response.setHeader("Cache-Control", "no-cache");
+
+ final AsyncContext asyncContext = request.startAsync();
+ asyncContext.setTimeout(0);
+
+ Endpoint endpoint = action.getEndpoint();
+
+ Runnable[] disposeSseListener = {null};
+
+ // Detect when client disconnects
+ asyncContext.addListener(new AsyncListener() {
+ @Override
+ public void onComplete(AsyncEvent event) {
+ disposeSseListener[0].run();
+ }
+
+ @Override
+ public void onTimeout(AsyncEvent event) {
+ disposeSseListener[0].run();
+ }
+
+ @Override
+ public void onError(AsyncEvent event) {
+ disposeSseListener[0].run();
+ }
+
+ @Override
+ public void onStartAsync(AsyncEvent event) {
+ // No-op
+ }
+ });
+
+ disposeSseListener[0] = () -> {
+ listenersByEndpoint.compute(endpoint, (et, clts) -> {
+ EndpointClients r = clts;
+ if (clts != null) {
+ // Remove the listener for the async context.
+ // If no more listeners remain then dispose the exec tracker listener.
+ clts.eventListeners.remove(asyncContext);
+ if (clts.eventListeners.isEmpty()) {
+ r = null;
+ }
+ return r;
+ }
+ return r;
+ });
+ };
+
+ // Atomically set up the new listener.
+ listenersByEndpoint.compute(endpoint, (et, clients) -> {
+ if (clients == null) {
+ clients = new EndpointClients();
+ }
+ clients.eventListeners.put(asyncContext, disposeSseListener[0]);
+ return clients;
+ });
+ }
+
+ /**
+ * Remove all graphs from the index for which there is no corresponding graph in the dataset.
+ */
+ protected void doClean(HttpAction action) throws Exception {
+ DatasetGraph dsg = action.getDataset();
+ Endpoint endpoint = action.getEndpoint();
+
+ TaskListener<BasicTask> taskListener = task -> {
+ switch (task.getTaskState()) {
+ case STARTING: {
+ JsonObject json = toJsonTaskStart(task.getStartTime(), null);
+ broadcastJson(endpoint, json);
+ break;
+ }
+ case TERMINATED: {
+ JsonObject json = toJsonTaskEnd(task.getEndTime(), task.getThrowable(), task.getStatusMessage());
+ broadcastJson(endpoint, json);
+ break;
+ }
+ default:
+ break;
+ }
+ };
+
+ try {
+ SpatialIndexLib.scheduleOnceCleanTask(dsg, taskListener);
+ } catch (Exception e) {
+ ServletOps.errorOccurred(e.getMessage(), e);
+ }
+ }
+
+ /** Send a stop request to a running task. Does not wait for the task to terminate. */
+ protected void doCancel(HttpAction action) {
+ BasicTask task = getActiveTask(action);
+ String state;
+ if (task != null) {
+ state = "true";
+ task.abort();
+ } else {
+ state = "false";
+ }
+
+ String jsonStr = String.format("{ \"stopped\": %s }", state);
+ successJson(action, jsonStr);
+ }
+
+ /**
+ * Serves a JSON object that captures the current indexing status.
+ *
+ * <pre>
+ * {
+ * isIndexing: Boolean, // Whether an index task is running.
+ * isAborting: Boolean, // Iff isIndexing==true: Whether abort has been requested.
+ * error: String, // Iff isIndexing==false: An error message of the most recently terminated indexing task.
+ * message: String // Status message of the most recent indexing task (running or terminated).
+ * time: long // EpochMillis of the last status update.
+ * }
+ * </pre>
+ */
+ protected void serveStatus(HttpAction action) {
+ BasicTask task = getActiveTask(action);
+
+ JsonObject status = new JsonObject();
+ long time;
+ if (task == null) {
+ status.addProperty("isIndexing", false);
+ time = 0;
+ } else {
+ status.addProperty("isIndexing", !task.isTerminated());
+ if (!task.isTerminated()) {
+ status.addProperty("isAborting", task.isAborting());
+ time = !task.isAborting() ? task.getStartTime() : task.getAbortTime();
+ } else {
+ time = task.getEndTime();
+ }
+ Throwable throwable = task.getThrowable();
+ if (throwable != null) {
+ String msg = ExceptionUtils.getStackTrace(throwable);
+ status.addProperty("error", msg);
+ }
+
+ String msg = task.getStatusMessage();
+ if (msg != null) {
+ status.addProperty("message", msg);
+ }
+ }
+ status.addProperty("time", time);
+
+
+ String jsonStr = gsonForSse.toJson(status);
+ successJson(action, jsonStr);
+ }
+
+ protected BasicTask scheduleIndexTask(HttpAction action, SpatialIndexerComputation indexComputation, Path targetFile, boolean isReplaceTask) {
+ Endpoint endpoint = action.getEndpoint();
+ DatasetGraph dsg = action.getDataset();
+
+ long graphCount = indexComputation.getGraphNodes().size();
+
+ TaskListener<BasicTask> taskListener = new TaskListener<BasicTask>() {
+ @Override
+ public void onStateChange(BasicTask task) {
+ switch (task.getTaskState()) {
+ case STARTING: {
+ JsonObject json = toJsonTaskStart(task.getStartTime(), null);
+ broadcastJson(endpoint, json);
+ break;
+ }
+ case ABORTING: {
+ JsonObject json = toJsonTaskAbort(task.getAbortTime(), null);
+ broadcastJson(endpoint, json);
+ break;
+ }
+ case TERMINATED: {
+ Throwable throwable = task.getThrowable();
+ long endTime = task.getEndTime();
+ JsonObject json = toJsonTaskEnd(endTime, throwable, task.getStatusMessage());
+ broadcastJson(endpoint, json);
+ if (logger.isInfoEnabled()) {
+ logger.info("Indexing task of {} graphs terminated.", graphCount);
+ }
+ }
+ default:
+ break;
+ }
+ }
+ };
+
+ return SpatialIndexLib.scheduleOnceIndexTask(dsg, indexComputation, targetFile, isReplaceTask, taskListener);
+ }
+
+ protected void doIndex(HttpAction action) throws Exception {
+ DatasetGraph dsg = action.getDataset();
+ SpatialIndex index = SpatialIndexLib.getSpatialIndex(dsg.getContext());
+
+ if (index == null) { // error: no spatial index has been configured
+ // XXX Could still allow for creating an ad-hoc in-memory-only index.
+ String msg = String.format("[%d] No spatial index has been configured for the dataset", action.id);
+ action.log.error(msg);
+ ServletOps.error(HttpSC.SERVICE_UNAVAILABLE_503, msg);
+ } else {
+ boolean isReplaceMode = isReplaceMode(action);
+ int threadCount = getThreadCount(action);
+
+ // Only SpatialIndexPerGraph can be updated.
+ // Check if the index can be updated.
+ // If not then raise an exception
+ // that informs that only replace mode can be used in this situation.
+ if (!(index instanceof SpatialIndexPerGraph)) {
+ boolean isUpdateMode = !isReplaceMode;
+ if (isUpdateMode) {
+ throw new RuntimeException("Cannot update existing spatial index because its type is unsupported. Consider replacing the index.");
+ }
+ }
+
+ Path oldLocation = index.getLocation();
+ if (oldLocation == null) {
+ action.log.warn("Spatial index will not be persisted because no file location was configured.");
+ }
+
+ String srsURI = index.getSrsInfo().getSrsURI();
+
+ List<Node> graphNodes = new ArrayList<>(Txn.calculateRead(dsg, () -> extractGraphsFromRequest(dsg, action)));
+ SpatialIndexerComputation task = new SpatialIndexerComputation(dsg, srsURI, graphNodes, threadCount);
+
+ action.log.info(String.format("[%d] spatial index: computation request accepted.", action.id));
+
+ try {
+ scheduleIndexTask(action, task, oldLocation, isReplaceMode);
+ successText(action, "Spatial index computation task accepted at " + DateTimeUtils.nowAsXSDDateTimeString());
+ } catch (Exception e) {
+ ServletOps.errorOccurred(e.getMessage(), e);
+ }
+ }
+ }
+
+ protected static JsonObject toJsonTaskStart(long timeInMillis, String msg) {
+ JsonObject json = new JsonObject();
+ json.addProperty("isIndexing", true);
+ json.addProperty("time", timeInMillis);
+ if (msg != null) {
+ json.addProperty("message", msg);
+ }
+ return json;
+ }
+ protected static JsonObject toJsonTaskAbort(long timeInMillis, String msg) {
+ JsonObject json = new JsonObject();
+ json.addProperty("isIndexing", true);
+ json.addProperty("isAborting", true);
+ json.addProperty("time", timeInMillis);
+ if (msg != null) {
+ json.addProperty("message", msg);
+ }
+ return json;
+ }
+
+ protected static JsonObject toJsonTaskEnd(long timeInMillis, Throwable throwable, String msg) {
+ JsonObject json = new JsonObject();
+ json.addProperty("isIndexing", false);
+ json.addProperty("time", timeInMillis);
+ if (msg != null) {
+ json.addProperty("message", msg);
+ }
+ if (throwable != null) {
+ json.addProperty("error", ExceptionUtils.getStackTrace(throwable));
+ }
+ return json;
+ }
+
+ protected void broadcastJson(Endpoint endpoint, JsonElement jsonData) {
+ EndpointClients clients = listenersByEndpoint.get(endpoint);
+ if (clients != null) {
+ Iterator<Entry<AsyncContext, Runnable>> it = clients.eventListeners.entrySet().iterator();
+ broadcastJson(it, jsonData);
+ }
+ }
+
+ protected void broadcastJson(Iterator<Entry<AsyncContext, Runnable>> it, JsonElement jsonData) {
+ String str = gsonForSse.toJson(jsonData);
+ broadcastLine(it, str);
+ }
+
+ /**
+ * Broadcast a payload to all listeners in the given iterator.
+ * On failure, the listener is removed from the iterator.
+ */
+ protected void broadcastLine(Iterator<Entry<AsyncContext, Runnable>> it, String payload) {
+ while (it.hasNext()) {
+ Entry<AsyncContext, Runnable> e = it.next();
+ AsyncContext context = e.getKey();
+ Runnable unregister = e.getValue();
+ try {
+ PrintWriter writer = context.getResponse().getWriter();
+ // Format demanded by server side events is: "data: <payload>\n\n".
+ writer.println("data: " + payload);
+ writer.println();
+ writer.flush();
+ } catch (Throwable x) {
+ it.remove(); // Remove first so that unregister does not cause concurrent modification.
+ logger.warn("Broadcast failed.", x);
+ try {
+ unregister.run();
+ } finally {
+ context.complete();
+ }
+ }
+ }
+ }
+
+ protected static void successText(HttpAction action, String jsonStr) {
+ successStringUtf8(action, WebContent.contentTypeTextPlain, jsonStr);
+ }
+
+ protected static void successJson(HttpAction action, String jsonStr) {
+ successStringUtf8(action, WebContent.contentTypeJSON, jsonStr);
+ }
+
+ protected static void successStringUtf8(HttpAction action, String contentType, String str) {
+ action.setResponseContentType(contentType);
+ action.setResponseCharacterEncoding(WebContent.charsetUTF8);
+ action.setResponseStatus(HttpSC.OK_200);
+ try {
+ action.getResponseOutputStream().println(str);
+ } catch (IOException e) {
+ IO.exception(e);
+ }
+ return;
+ }
+}
diff --git a/jena-fuseki2/jena-fuseki-mod-geosparql/src/main/java/org/apache/jena/fuseki/mod/geosparql/TaskTracker.java b/jena-fuseki2/jena-fuseki-mod-geosparql/src/main/java/org/apache/jena/fuseki/mod/geosparql/TaskTracker.java
deleted file mode 100644
index 93f15dff68..0000000000
--- a/jena-fuseki2/jena-fuseki-mod-geosparql/src/main/java/org/apache/jena/fuseki/mod/geosparql/TaskTracker.java
+++ /dev/null
@@ -1,128 +0,0 @@
-/*
- * 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.jena.fuseki.mod.geosparql;
-
-import java.time.Duration;
-import java.time.Instant;
-import java.util.Iterator;
-import java.util.Map.Entry;
-import java.util.Objects;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.ConcurrentSkipListMap;
-import java.util.concurrent.atomic.AtomicLong;
-
-import org.apache.jena.sparql.SystemARQ;
-import org.apache.jena.sparql.util.Context;
-import org.apache.jena.sparql.util.Symbol;
-
-/**
- * A simple tracker for abortable tasks.
- * Task registration accepts an arbitrary object and a runnable for aborting the task.
- * The task itself is responsible for deregistration on its termination.
- * A record with starting and termination time is maintained by this class.
- */
-public class TaskTracker {
- /** Record for when a task is submitted to the tracker. */
- public record StartRecord(long requestId, Instant timestamp, Object requestObject, Runnable abortAction) {}
-
- /** Record for when a task is completed. */
- public record CompletionRecord(StartRecord start, Instant timestamp, Throwable throwable) {
- public Duration duration() {
- return Duration.between(start.timestamp, timestamp);
- }
-
- public boolean isSuccess() {
- return throwable == null;
- }
- }
-
- protected AtomicLong nextId = new AtomicLong();
- protected ConcurrentMap<Long, StartRecord> idToStartRecord = new ConcurrentHashMap<>();
- protected int maxHistorySize = 1000;
- protected ConcurrentMap<Instant, CompletionRecord> history = new ConcurrentSkipListMap<>();
-
- public ConcurrentMap<Instant, CompletionRecord> getHistory() {
- return history;
- }
-
- public void setMaxHistorySize(int maxHistorySize) {
- this.maxHistorySize = maxHistorySize;
- }
-
- public long put(Object requestObject, Runnable abortAction) {
- long result = nextId.getAndIncrement();
- StartRecord record = new StartRecord(result, Instant.now(), requestObject, abortAction);
- idToStartRecord.put(result, record);
- return result;
- }
-
- protected void trimHistory() {
- if (history.size() >= maxHistorySize) {
- Iterator<Entry<Instant, CompletionRecord>> it = history.entrySet().iterator();
- while (history.size() >= maxHistorySize && it.hasNext()) {
- it.next();
- it.remove();
- }
- }
- }
-
- public CompletionRecord remove(long id, Throwable t) {
- StartRecord startRecord = idToStartRecord.remove(id);
- CompletionRecord result = null;
- if (startRecord != null) {
- trimHistory();
- Instant now = Instant.now();
- result = new CompletionRecord(startRecord, now, t);
- history.put(now, result);
- }
- return result;
- }
-
- @Override
- public String toString() {
- return "Active: " + idToStartRecord.size() + ", History: " + history.size() + "/" + maxHistorySize;
- }
-
- public static final Symbol symTaskTracker = SystemARQ.allocSymbol("taskTracker");
-
- public static TaskTracker getTracker(Context context) {
- return context.get(symTaskTracker);
- }
-
- public static TaskTracker requireTracker(Context context) {
- TaskTracker result = getTracker(context);
- Objects.requireNonNull("No ExecTracker registered in context");
- return result;
- }
-
- public static TaskTracker ensureTracker(Context context) {
- TaskTracker result = context.get(symTaskTracker);
- if (result == null) {
- synchronized (context) {
- result = context.get(symTaskTracker);
- if (result == null) {
- result = new TaskTracker();
- context.set(symTaskTracker, result);
- }
- }
- }
- return result;
- }
-}
diff --git a/jena-fuseki2/jena-fuseki-mod-geosparql/src/main/resources/spatial-indexer/index.html b/jena-fuseki2/jena-fuseki-mod-geosparql/src/main/resources/spatial-indexer/index.html
index 117c0ea9f5..be6caaacfb 100644
--- a/jena-fuseki2/jena-fuseki-mod-geosparql/src/main/resources/spatial-indexer/index.html
+++ b/jena-fuseki2/jena-fuseki-mod-geosparql/src/main/resources/spatial-indexer/index.html
@@ -86,7 +86,6 @@
<div style="display:flex; justify-content:flex-end; margin-left: auto;">
<button id="clean-action">Remove absent graphs from index</button>
</div>
- <!-- <button id="stop-action">Stop running action</button> -->
</div>
<!--
@@ -113,7 +112,7 @@
</table>
<script>
- const sparqlEndpoint = document.location.href + '/../';
+ // const sparqlEndpoint = document.location.href + '/../';
const apiEndpoint = document.location.href;
const eventEndpoint = new EventSource(apiEndpoint + '?command=events');
@@ -153,7 +152,7 @@
}
}
- async function cancelIndexing() {
+ async function cancelTask() {
const params = new URLSearchParams({ "command": "cancel" });
try {
const response = await fetch(apiEndpoint, { "method": "POST", "body": params })
@@ -167,6 +166,21 @@
}
async function fetchGraphs() {
+ const url = apiEndpoint + '?command=graphs';
+
+ try {
+ const response = await fetch(url, { "method": "POST" });
+ const data = await response.json();
+ return data;
+ // return data.results.bindings.map(binding => binding.g.value);
+ } catch (error) {
+ console.error('Error fetching graphs:', error);
+ return [];
+ }
+ }
+
+ // No longer used in order to make spatial indexer endpoint self-contained!
+ async function fetchGraphsViaSparql() {
const query = `SELECT ?g { { BIND(<urn:x-arq:DefaultGraph> AS ?g) } UNION { BIND(<urn:x-arq:UnionGraph> AS ?g) } UNION { { SELECT * { GRAPH ?g { } } ORDER BY ?g } } }`;
const url = `${sparqlEndpoint}?query=${encodeURIComponent(query)}&format=application/json`;
@@ -221,6 +235,7 @@
const deselectAllBtn = document.getElementById('deselect-all');
const applyBtn = document.getElementById('apply-action');
+ const stopBtn = document.getElementById('stop-action');
const cleanBtn = document.getElementById('clean-action');
const replaceCb = document.getElementById('replace-toggle');
const maxThreadCountInput = document.getElementById('max-thread-count');
@@ -275,6 +290,10 @@
reIndexGraphs(selectedGraphs);
});
+ stopBtn.addEventListener('click', function() {
+ cancelTask();
+ });
+
cleanBtn.addEventListener('click', function () {
cleanGraphs();
});
@@ -290,7 +309,7 @@
}
function updateCancelButton() {
- const elt = document.getElementById('stop-action');
+ const elt = stopBtn;
elt.style.display = 'none';
elt.innerHTML = "Stop running indexer.";
elt.disabled = false;
@@ -353,14 +372,6 @@
}
}
- document.getElementById('stop-action').addEventListener('click', function() {
- cancelIndexing();
- });
-
- cleanBtn.addEventListener('click', function() {
- cancelIndexing();
- });
-
async function initialize() {
status = await fetchIndexerStatus();
updateStatus();
diff --git a/jena-fuseki2/jena-fuseki-mod-geosparql/src/test/java/org/apache/jena/fuseki/mod/geosparql/TestFMod_SpatialIndexer.java b/jena-fuseki2/jena-fuseki-mod-geosparql/src/test/java/org/apache/jena/fuseki/mod/geosparql/TestFMod_SpatialIndexer.java
index 391814a4db..30e0a449a5 100644
--- a/jena-fuseki2/jena-fuseki-mod-geosparql/src/test/java/org/apache/jena/fuseki/mod/geosparql/TestFMod_SpatialIndexer.java
+++ b/jena-fuseki2/jena-fuseki-mod-geosparql/src/test/java/org/apache/jena/fuseki/mod/geosparql/TestFMod_SpatialIndexer.java
@@ -31,8 +31,8 @@ import org.apache.jena.geosparql.spatial.SpatialIndex;
import org.apache.jena.geosparql.spatial.SpatialIndexException;
import org.apache.jena.geosparql.spatial.index.v2.GeometryGenerator;
import org.apache.jena.geosparql.spatial.index.v2.GeometryGenerator.GeometryType;
+import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexLib;
import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexPerGraph;
-import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexUtils;
import org.apache.jena.graph.Node;
import org.apache.jena.graph.NodeFactory;
import org.apache.jena.sparql.core.DatasetGraph;
@@ -76,16 +76,19 @@ public class TestFMod_SpatialIndexer {
dsg = DatasetGraphFactory.create();
setupTestData(dsg);
- spatialIndex = SpatialIndexUtils.buildSpatialIndex(dsg);
+ spatialIndex = SpatialIndexLib.buildSpatialIndex(dsg);
String[] argv = new String[] { "--empty" };
+
FusekiServer server = FusekiMain.builder(argv)
.add("test", dsg)
+ .registerOperation(FMod_SpatialIndexer.spatialIndexerOperation, new SpatialIndexerService())
+ .addEndpoint("test", "spatial-indexer", FMod_SpatialIndexer.spatialIndexerOperation)
.build();
server.start();
int port = server.getPort();
String serverURL = "http://localhost:" + port + "/";
- String siteUrl = serverURL + "test/update-spatial";
+ String siteUrl = serverURL + "test/spatial-indexer";
ChromeOptions options = new ChromeOptions();
options.addArguments("--headless=new"); // use "new" headless mode (better support)
@@ -132,7 +135,7 @@ public class TestFMod_SpatialIndexer {
WebElement cleanButton = driver.findElement(By.id("clean-action"));
cleanButton.click();
awaitEvent();
- SpatialIndexPerGraph newIndex = (SpatialIndexPerGraph)SpatialIndexUtils.getSpatialIndex(dsg.getContext());
+ SpatialIndexPerGraph newIndex = (SpatialIndexPerGraph)SpatialIndexLib.getSpatialIndex(dsg.getContext());
// Check the number of graphs in the spatial index; the treeMap includes the default graph.
long numGraphsInIndex = newIndex.getIndex().getTreeMap().keySet().size();
diff --git a/jena-fuseki2/jena-fuseki-mod-geosparql/src/test/java/org/apache/jena/fuseki/mod/geosparql/TestSpatialIndexerTasks.java b/jena-fuseki2/jena-fuseki-mod-geosparql/src/test/java/org/apache/jena/fuseki/mod/geosparql/TestSpatialIndexerTasks.java
new file mode 100644
index 0000000000..29c91a725d
--- /dev/null
+++ b/jena-fuseki2/jena-fuseki-mod-geosparql/src/test/java/org/apache/jena/fuseki/mod/geosparql/TestSpatialIndexerTasks.java
@@ -0,0 +1,209 @@
+/*
+ * 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.jena.fuseki.mod.geosparql;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.function.Predicate;
+
+import org.apache.jena.geosparql.implementation.vocabulary.SRS_URI;
+import org.apache.jena.geosparql.spatial.SpatialIndexException;
+import org.apache.jena.geosparql.spatial.index.v2.GeometryGenerator;
+import org.apache.jena.geosparql.spatial.index.v2.GeometryGenerator.GeometryType;
+import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexPerGraph;
+import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexLib;
+import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexerComputation;
+import org.apache.jena.graph.Graph;
+import org.apache.jena.graph.Node;
+import org.apache.jena.graph.NodeFactory;
+import org.apache.jena.sparql.core.DatasetGraph;
+import org.apache.jena.sparql.core.DatasetGraphFactory;
+import org.apache.jena.sparql.core.Quad;
+import org.junit.Assert;
+import org.junit.Test;
+import org.locationtech.jts.geom.Envelope;
+
+/** Test cases that check for whether the correct graphs are indexed - also when a user is only authorized to a certain subset of graphs. */
+public class TestSpatialIndexerTasks {
+
+ private static final Node graph(int i) {
+ return NodeFactory.createURI("http://www.example.org/graph" + i);
+ }
+
+ private static final Node dg = Quad.defaultGraphIRI;
+ private static final Node g1 = graph(1);
+ private static final Node g2 = graph(2);
+ private static final Node g3 = graph(3);
+ private static final Node g4 = graph(4);
+
+ private static final Set<Node> allGraphs = Set.of(Quad.defaultGraphIRI, g1, g2, g3, g4);
+
+ /** Test data: spatial data in 1 default graph (dg) and 4 named graphs (g1, g2, g3, 4) */
+ private static DatasetGraph createTestData() {
+ Map<GeometryType, Number> conf = GeometryGenerator.createConfig(1);
+ DatasetGraph dsg = DatasetGraphFactory.create();
+ for (int i = 0; i < 5; ++i) {
+ Node graph = i == 0
+ ? dg
+ : graph(i);
+ Graph g = dsg.getGraph(graph);
+ Envelope env = new Envelope(-175, 175, -85, 85);
+ GeometryGenerator.generateGraph(g, env, conf);
+ }
+ return dsg;
+ }
+
+ /** Helper method: Replace a prior index with the given one. Returns the new spatial index. */
+ private static SpatialIndexPerGraph replaceSpatialIndex(DatasetGraph dsg, Predicate<Node> isAuthorizedGraph, String srs, Collection<Node> replaceGraphs) {
+ List<Node> graphNodes = List.copyOf(replaceGraphs);
+ SpatialIndexerComputation computation = new SpatialIndexerComputation(dsg, srs, graphNodes, 1);
+ SpatialIndexLib.createIndexerTask(dsg, isAuthorizedGraph, computation, null, null, true).run();
+ SpatialIndexPerGraph spatialIndex = SpatialIndexLib.getSpatialIndex(dsg);
+ SpatialIndexLib.setSpatialIndex(dsg, spatialIndex);
+
+ // Access the spatial index via context to test the machinery.
+ SpatialIndexPerGraph result = SpatialIndexLib.getSpatialIndex(dsg);
+ return result;
+ }
+
+ /** Helper method: Get or create a spatial index and update the requested graphs in it. Returns the new spatial index. */
+ private static SpatialIndexPerGraph updateSpatialIndex(DatasetGraph dsg, Predicate<Node> isAuthorizedGraph, String srs, Collection<Node> updateGraphs) {
+ List<Node> graphNodes = List.copyOf(updateGraphs);
+ SpatialIndexerComputation computation = new SpatialIndexerComputation(dsg, srs, graphNodes, 1);
+ SpatialIndexLib.createIndexerTask(dsg, isAuthorizedGraph, computation, null, null, false).run();
+ SpatialIndexPerGraph spatialIndex = SpatialIndexLib.getSpatialIndex(dsg);
+ return spatialIndex;
+ }
+
+ /** Helper method: run a clean task and return the resulting index from the context. */
+ private static SpatialIndexPerGraph clean(DatasetGraph dsg, Predicate<Node> isAuthorizedGraph) {
+ SpatialIndexLib.createCleanTask(dsg, isAuthorizedGraph, null).run();
+ SpatialIndexPerGraph spatialIndex = SpatialIndexLib.getSpatialIndex(dsg);
+ return spatialIndex;
+ }
+
+ @Test
+ public void testDsgIndexUpdate() throws SpatialIndexException {
+ DatasetGraph dsg = createTestData();
+
+ Set<Node> initialGraphs = Set.of(g2, g3, g4);
+ Set<Node> updateGraphs = Set.of(dg, g1);
+
+ replaceSpatialIndex(dsg, null, SRS_URI.DEFAULT_WKT_CRS84, initialGraphs);
+ SpatialIndexPerGraph spatialIndex = updateSpatialIndex(dsg, null, SRS_URI.DEFAULT_WKT_CRS84, updateGraphs);
+
+ Set<Node> indexedGraphNodes = spatialIndex.getIndex().getTreeMap().keySet();
+ Assert.assertEquals(allGraphs, indexedGraphNodes);
+ }
+
+ @Test
+ public void testDsgIndexReplace() throws SpatialIndexException {
+ DatasetGraph dsg = createTestData();
+
+ SpatialIndexLib.buildSpatialIndex(dsg, SRS_URI.DEFAULT_WKT_CRS84);
+
+ Set<Node> expectedGraphSet = Set.of(g1, g2);
+ SpatialIndexPerGraph spatialIndex = replaceSpatialIndex(dsg, null, SRS_URI.DEFAULT_WKT_CRS84, expectedGraphSet);
+
+ Set<Node> actualGraphSet = spatialIndex.getIndex().getTreeMap().keySet();
+ Assert.assertEquals(expectedGraphSet, actualGraphSet);
+ }
+
+ @Test
+ public void testDsgIndexClean() throws InterruptedException, ExecutionException, SpatialIndexException {
+ DatasetGraph dsg = createTestData();
+
+ SpatialIndexLib.buildSpatialIndex(dsg, SRS_URI.DEFAULT_WKT_CRS84);
+
+ Set.of(dg, g1, g2, g3).forEach(dsg::removeGraph);
+
+ SpatialIndexPerGraph spatialIndex = clean(dsg, null);
+ Set<Node> actual = spatialIndex.getIndex().getTreeMap().keySet();
+ Set<Node> expected = Set.of(dg, g4);;
+ Assert.assertEquals(expected, actual);
+ }
+
+ @Test
+ public void testProtectedDsgIndexUpdate() throws SpatialIndexException {
+ DatasetGraph dsg = createTestData();
+
+ // User can only see g3 and g4 and requests to update both of them.
+ Set<Node> authorizedGraph = Set.of(g3, g4);
+ Set<Node> initialGraphs = Set.of(dg, g1, g2, g3);
+ Set<Node> updateGraphs = Set.of(g3, g4);
+
+ replaceSpatialIndex(dsg, null, SRS_URI.DEFAULT_WKT_CRS84, initialGraphs);
+ SpatialIndexPerGraph spatialIndex = updateSpatialIndex(dsg, authorizedGraph::contains, SRS_URI.DEFAULT_WKT_CRS84, updateGraphs);
+
+ Set<Node> indexedGraphNodes = spatialIndex.getIndex().getTreeMap().keySet();
+ Assert.assertEquals(allGraphs, indexedGraphNodes);
+ }
+
+ @Test
+ public void testProtectedDsgIndexReplace() {
+ DatasetGraph dsg = createTestData();
+
+ // User can only see g3 and g4 and requests to replace index with only g4 - so g3 should get lost.
+ Set<Node> authorizedGraphs = Set.of(g3, g4);
+
+ Set<Node> initialGraphs = Set.of(dg, g1, g2, g3);
+ Set<Node> replaceGraphs = Set.of(g4);
+ Set<Node> expectedGraphs = Set.of(dg, g1, g2, g4);
+
+ replaceSpatialIndex(dsg, null, SRS_URI.DEFAULT_WKT_CRS84, initialGraphs);
+ SpatialIndexPerGraph spatialIndex = replaceSpatialIndex(dsg, authorizedGraphs::contains, SRS_URI.DEFAULT_WKT_CRS84, replaceGraphs);
+
+ Set<Node> indexedGraphNodes = spatialIndex.getIndex().getTreeMap().keySet();
+ Assert.assertEquals(expectedGraphs, indexedGraphNodes);
+ }
+
+ @Test
+ public void testProtectedDsgCleanAuthorized() throws SpatialIndexException {
+ DatasetGraph dsg = createTestData();
+
+ // User is authorized for g3 and g4 - but g4 gets removed -> g4 should become removed.
+ Set<Node> authorizedGraphs = Set.of(g3, g4);
+
+ SpatialIndexLib.buildSpatialIndex(dsg, SRS_URI.DEFAULT_WKT_CRS84);
+ Set.of(dg, g4).forEach(dsg::removeGraph);
+
+ SpatialIndexPerGraph spatialIndex = clean(dsg, authorizedGraphs::contains);
+ Set<Node> actual = spatialIndex.getIndex().getTreeMap().keySet();
+ Set<Node> expected = Set.of(dg, g1, g2, g3);
+ Assert.assertEquals(expected, actual);
+ }
+
+ @Test
+ public void testProtectedDsgCleanUnauthorized() throws SpatialIndexException {
+ DatasetGraph dsg = createTestData();
+ SpatialIndexLib.buildSpatialIndex(dsg, SRS_URI.DEFAULT_WKT_CRS84);
+ Set.of(dg, g1, g3).forEach(dsg::removeGraph);
+
+ // Among the authorized graphs, only g4 has been removed and can thus be cleaned up.
+ Set<Node> authorizedGraphs = Set.of(g3, g4);
+
+ SpatialIndexPerGraph spatialIndex = clean(dsg, authorizedGraphs::contains);
+ Set<Node> actual = spatialIndex.getIndex().getTreeMap().keySet();
+ Set<Node> expected = Set.of(dg, g1, g2, g4);
+ Assert.assertEquals(expected, actual);
+ }
+}
diff --git a/jena-fuseki2/jena-fuseki-server/src/main/java/org/apache/jena/playground/MainPlaygroundFusekiModGeoSparql.java b/jena-fuseki2/jena-fuseki-server/src/main/java/org/apache/jena/playground/MainPlaygroundFusekiModGeoSparql.java
deleted file mode 100644
index 2e2518ea09..0000000000
--- a/jena-fuseki2/jena-fuseki-server/src/main/java/org/apache/jena/playground/MainPlaygroundFusekiModGeoSparql.java
+++ /dev/null
@@ -1,46 +0,0 @@
-/*
- * 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.jena.playground;
-
-import org.apache.jena.fuseki.main.FusekiServer;
-import org.apache.jena.fuseki.main.sys.FusekiModules;
-import org.apache.jena.fuseki.mod.geosparql.FMod_SpatialIndexer;
-import org.apache.jena.fuseki.system.FusekiLogging;
-import org.apache.jena.query.DatasetFactory;
-
-public class MainPlaygroundFusekiModGeoSparql {
- public static void main(String[] args) {
- FusekiLogging.setLogging();
-
- FMod_SpatialIndexer myModule = new FMod_SpatialIndexer();
- myModule.start();
- FusekiModules fmods = FusekiModules.create(myModule);
- FusekiServer server = FusekiServer.create()
- .fusekiModules(fmods)
- .staticFileBase("/home/raven/Laboratory/webapp")
- .add("test", DatasetFactory.create())
- .port(3030)
- .enablePing(true)
- .enableStats(true)
- .enableTasks(true)
- .build();
-
- server.start();
- }
-}
diff --git a/jena-geosparql/src/main/java/org/apache/jena/geosparql/assembler/GeoAssembler.java b/jena-geosparql/src/main/java/org/apache/jena/geosparql/assembler/GeoAssembler.java
index 0f64669160..213b4c5b6d 100644
--- a/jena-geosparql/src/main/java/org/apache/jena/geosparql/assembler/GeoAssembler.java
+++ b/jena-geosparql/src/main/java/org/apache/jena/geosparql/assembler/GeoAssembler.java
@@ -41,8 +41,8 @@ import org.apache.jena.atlas.io.IO;
import org.apache.jena.geosparql.configuration.GeoSPARQLConfig;
import org.apache.jena.geosparql.configuration.GeoSPARQLOperations;
import org.apache.jena.geosparql.configuration.SrsException;
+import org.apache.jena.geosparql.spatial.SpatialIndexConstants;
import org.apache.jena.geosparql.spatial.SpatialIndexException;
-import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexUtils;
import org.apache.jena.query.Dataset;
import org.apache.jena.query.DatasetFactory;
import org.apache.jena.rdf.model.Property;
@@ -122,15 +122,24 @@ public class GeoAssembler extends DatasetAssembler {
if (root.hasProperty(pSpatialIndexFile) )
spatialIndexFilename = GraphUtils.getStringValue(root, pSpatialIndexFile);
+ if (spatialIndexFilename == null) {
+ LOG.warn(root + ": No spatial index file location is specified via " + pSpatialIndexFile + ". Spatial index will not be persisted.");
+ }
+
// SRS URI
String srsURI = null;
- if (root.hasProperty(pSrsUri) )
- srsURI = GraphUtils.getStringValue(root, pSrsUri);
+ if (root.hasProperty(pSrsUri) ) {
+ srsURI = GraphUtils.getResourceValue(root, pSrsUri).getURI();
+ }
+
+ if (srsURI == null) {
+ LOG.warn(root + ": No preferred SRS is configured via " + pSrsUri);
+ }
// ---- Build
Dataset dataset = DatasetFactory.wrap(base);
- dataset.getContext().set(SpatialIndexUtils.symSrsUri, srsURI);
+ dataset.getContext().set(SpatialIndexConstants.symSrsUri, srsURI);
// Conversion of data. Startup-only.
// needed for w3c:geo/wgs84_pos#lat/log.
@@ -178,13 +187,9 @@ public class GeoAssembler extends DatasetAssembler {
private static void prepareSpatialExtension(Dataset dataset, String spatialIndex){
boolean isEmpty = dataset.calculateRead(dataset::isEmpty);
- if ( isEmpty && spatialIndex != null ) {
- LOG.warn("Dataset empty. Spatial Index not constructed. Server will require restarting after adding data and any updates to build Spatial Index.");
- return;
+ if ( isEmpty ) {
+ LOG.warn("Dataset is empty. Constructing an empty spatial index that needs to be updated once data is added.");
}
- if ( isEmpty )
- // Nothing to do.
- return;
try {
// no file given, i.e. in-memory index only
diff --git a/jena-geosparql/src/main/java/org/apache/jena/geosparql/configuration/GeoSPARQLConfig.java b/jena-geosparql/src/main/java/org/apache/jena/geosparql/configuration/GeoSPARQLConfig.java
index 4f61fe1cd2..dfc5c7aa42 100644
--- a/jena-geosparql/src/main/java/org/apache/jena/geosparql/configuration/GeoSPARQLConfig.java
+++ b/jena-geosparql/src/main/java/org/apache/jena/geosparql/configuration/GeoSPARQLConfig.java
@@ -39,7 +39,7 @@ import org.apache.jena.geosparql.spatial.SpatialIndex;
import org.apache.jena.geosparql.spatial.SpatialIndexException;
import org.apache.jena.geosparql.spatial.index.compat.SpatialIndexIo;
import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexIoKryo;
-import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexUtils;
+import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexLib;
import org.apache.jena.query.Dataset;
import org.apache.jena.sparql.function.FunctionRegistry;
import org.apache.jena.sparql.pfunction.PropertyFunctionRegistry;
@@ -258,7 +258,7 @@ public class GeoSPARQLConfig {
* @throws SpatialIndexException
*/
public static final void setupSpatialIndex(Dataset dataset) throws SpatialIndexException {
- SpatialIndexUtils.buildSpatialIndex(dataset.asDatasetGraph());
+ SpatialIndexLib.buildSpatialIndex(dataset.asDatasetGraph());
}
/**
@@ -271,7 +271,7 @@ public class GeoSPARQLConfig {
*/
public static final void setupPrecomputedSpatialIndex(Dataset dataset, Path spatialIndexFile) throws SpatialIndexException {
SpatialIndex si = SpatialIndexIo.load(spatialIndexFile);
- SpatialIndexUtils.setSpatialIndex(dataset, si);
+ SpatialIndexLib.setSpatialIndex(dataset, si);
}
/**
@@ -284,13 +284,9 @@ public class GeoSPARQLConfig {
* @throws SpatialIndexException
*/
public static final void setupSpatialIndex(Dataset dataset, Path spatialIndexFile) throws SpatialIndexException {
- SpatialIndexIoKryo.buildSpatialIndex(dataset, spatialIndexFile);
+ SpatialIndexIoKryo.loadOrBuildSpatialIndex(dataset, spatialIndexFile);
}
-// public static final void setupSpatialIndex(Dataset dataset, Path spatialIndexFile, boolean spatialIndexPerGraph) throws SpatialIndexException {
-// SpatialIndexIoKryo.buildSpatialIndex(dataset, spatialIndexFile, spatialIndexPerGraph);
-// }
-
/**
* Setup Spatial Index using Dataset using provided SRS URI.<br>
* Spatial Index written to file once created.
@@ -301,7 +297,7 @@ public class GeoSPARQLConfig {
* @throws SpatialIndexException
*/
public static final void setupSpatialIndex(Dataset dataset, String srsURI, Path spatialIndexFile) throws SpatialIndexException {
- SpatialIndexIoKryo.buildSpatialIndex(dataset, srsURI, spatialIndexFile);
+ SpatialIndexIoKryo.loadOrBuildSpatialIndex(dataset, srsURI, spatialIndexFile);
}
/**
diff --git a/jena-geosparql/src/main/java/org/apache/jena/geosparql/configuration/GeoSPARQLOperations.java b/jena-geosparql/src/main/java/org/apache/jena/geosparql/configuration/GeoSPARQLOperations.java
index 6c28630f13..4f5bc802f5 100644
--- a/jena-geosparql/src/main/java/org/apache/jena/geosparql/configuration/GeoSPARQLOperations.java
+++ b/jena-geosparql/src/main/java/org/apache/jena/geosparql/configuration/GeoSPARQLOperations.java
@@ -34,7 +34,9 @@ import java.util.List;
import java.util.Set;
import java.util.TreeSet;
import java.util.UUID;
+import java.util.stream.Stream;
+import org.apache.jena.atlas.iterator.Iter;
import org.apache.jena.datatypes.DatatypeFormatException;
import org.apache.jena.datatypes.RDFDatatype;
import org.apache.jena.geosparql.implementation.GeometryWrapper;
@@ -44,9 +46,11 @@ import org.apache.jena.geosparql.implementation.datatype.WKTDatatype;
import org.apache.jena.geosparql.implementation.index.GeometryLiteralIndex;
import org.apache.jena.geosparql.implementation.vocabulary.Geo;
import org.apache.jena.geosparql.implementation.vocabulary.GeoSPARQL_URI;
+import org.apache.jena.geosparql.implementation.vocabulary.SRS_URI;
import org.apache.jena.geosparql.implementation.vocabulary.SpatialExtension;
import org.apache.jena.geosparql.spatial.ConvertLatLon;
-import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexUtils;
+import org.apache.jena.geosparql.spatial.SpatialIndexConstants;
+import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexLib;
import org.apache.jena.query.Dataset;
import org.apache.jena.query.DatasetFactory;
import org.apache.jena.query.ReadWrite;
@@ -500,37 +504,57 @@ public class GeoSPARQLOperations {
}
/**
- * Find the most frequent SRS URI of Geometry Literals in the dataset.
- *
- * @param dataset
- * @return SRS URI
+ * Return an SRS for a dataset.
+ * If {@link SpatialIndexConstants#symSrsUri} is set then this SRS is returned.
+ * Otherwise, {@link #scanModeSRS(Dataset, String)} is called.
*/
public static final String findModeSRS(Dataset dataset) throws SrsException {
// return SRS if set via assembler config
- if (dataset.getContext().isDefined(SpatialIndexUtils.symSrsUri)) {
- return dataset.getContext().getAsString(SpatialIndexUtils.symSrsUri);
+ String result = SpatialIndexLib.getPreferredSRS(dataset.getContext());
+
+ if (result == null) {
+ LOGGER.info("SRS scan started. This may take a while...");
+
+ result = scanModeSRS(dataset, null);
+
+ if (result != null) {
+ LOGGER.info("SRS scan completed. Selecting SRS: " + result);
+ } else {
+ result = SRS_URI.DEFAULT_WKT_CRS84;
+ LOGGER.warn("SRS scan completed. No SRS usage found. Falling back to default SRS: " + result);
+ }
}
- LOGGER.info("Find Mode SRS - Started");
+ return result;
+ }
+
+ /**
+ * Find the most frequent SRS URI of Geometry Literals in the dataset.
+ *
+ * @param dataset
+ * @return SRS URI
+ */
+ public static final String scanModeSRS(Dataset dataset, String fallbackSrs) throws SrsException {
ModeSRS modeSRS = new ModeSRS();
- //Default Model
try (AutoTxn txn = Txn.autoTxn(dataset, ReadWrite.READ)) {
+ //Default Model
Model defaultModel = dataset.getDefaultModel();
modeSRS.search(defaultModel);
//Named Models
- Iterator<String> graphNames = dataset.listNames();
- while (graphNames.hasNext()) {
- String graphName = graphNames.next();
- Model namedModel = dataset.getNamedModel(graphName);
- modeSRS.search(namedModel);
+ try (Stream<String> graphNames = Iter.asStream(dataset.listNames())) {
+ Iterator<String> it = graphNames.iterator();
+ while (it.hasNext()) {
+ String graphName = it.next();
+ Model namedModel = dataset.getNamedModel(graphName);
+ modeSRS.search(namedModel);
+ }
}
-
txn.commit();
- LOGGER.info("Find Mode SRS - Completed");
}
-
- return modeSRS.getModeURI();
+ List<?> srsList = modeSRS.getSrsList();
+ String result = srsList.isEmpty() ? fallbackSrs : modeSRS.getModeURI();
+ return result;
}
/**
diff --git a/jena-geosparql/src/main/java/org/apache/jena/geosparql/geo/topological/GenericPropertyFunction.java b/jena-geosparql/src/main/java/org/apache/jena/geosparql/geo/topological/GenericPropertyFunction.java
index 4c0c827842..c1185a71ad 100644
--- a/jena-geosparql/src/main/java/org/apache/jena/geosparql/geo/topological/GenericPropertyFunction.java
+++ b/jena-geosparql/src/main/java/org/apache/jena/geosparql/geo/topological/GenericPropertyFunction.java
@@ -28,7 +28,7 @@ import org.apache.jena.geosparql.implementation.vocabulary.Geo;
import org.apache.jena.geosparql.implementation.vocabulary.SpatialExtension;
import org.apache.jena.geosparql.spatial.SpatialIndex;
import org.apache.jena.geosparql.spatial.SpatialIndexException;
-import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexUtils;
+import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexLib;
import org.apache.jena.graph.Graph;
import org.apache.jena.graph.Node;
import org.apache.jena.graph.Triple;
@@ -154,7 +154,7 @@ public abstract class GenericPropertyFunction extends PFuncSimple {
}
}
- boolean isSpatialIndex = SpatialIndexUtils.isDefined(execCxt);
+ boolean isSpatialIndex = SpatialIndexLib.isDefined(execCxt);
QueryIterator result;
if (!isSpatialIndex || filterFunction.isDisjoint() || filterFunction.isDisconnected()) {
//Disjointed so retrieve all cases.
@@ -230,12 +230,12 @@ public abstract class GenericPropertyFunction extends PFuncSimple {
Node geometryLiteral = boundGeometryLiteral.getGeometryLiteral();
// Perform the search of the Spatial Index of the Dataset.
- SpatialIndex spatialIndex = SpatialIndexUtils.retrieve(execCxt);
+ SpatialIndex spatialIndex = SpatialIndexLib.retrieve(execCxt);
GeometryWrapper geom = GeometryWrapper.extract(geometryLiteral);
GeometryWrapper transformedGeom = geom.transform(spatialIndex.getSrsInfo());
Envelope searchEnvelope = transformedGeom.getEnvelope();
- Node graphName = SpatialIndexUtils.unwrapGraphName(graph);
+ Node graphName = SpatialIndexLib.unwrapGraphName(graph);
Collection<Node> features = spatialIndex.query(searchEnvelope, graphName);
// Check each of the Features that match the search.
diff --git a/jena-geosparql/src/test/java/org/apache/jena/geosparql/spatial/TestSpatialIndexGraphLookupV2.java b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/SpatialIndexConstants.java
similarity index 62%
copy from jena-geosparql/src/test/java/org/apache/jena/geosparql/spatial/TestSpatialIndexGraphLookupV2.java
copy to jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/SpatialIndexConstants.java
index 508da2fc79..f9f9490d90 100644
--- a/jena-geosparql/src/test/java/org/apache/jena/geosparql/spatial/TestSpatialIndexGraphLookupV2.java
+++ b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/SpatialIndexConstants.java
@@ -15,16 +15,16 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.jena.geosparql.spatial;
-import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexUtils;
-import org.apache.jena.sparql.core.DatasetGraph;
+import org.apache.jena.sparql.util.Symbol;
+
+public class SpatialIndexConstants {
+
+ public static final Symbol symSpatialIndex = Symbol.create("http://jena.apache.org/spatial#index");
+ public static final Symbol symSrsUri = Symbol.create("http://jena.apache.org/spatial#srsURI");
-public class TestSpatialIndexGraphLookupV2
- extends AbstractSpatialIndexGraphLookpTest
-{
- @Override
- protected SpatialIndex buildSpatialIndex(DatasetGraph dsg, String srsUri) throws SpatialIndexException {
- return SpatialIndexUtils.buildSpatialIndex(dsg, srsUri);
- }
+ /** Symbol for a running task in a dataset's context. Used to synchronize tasks that modify the spatial index. */
+ public static final Symbol symSpatialIndexTask = Symbol.create("http://jena.apache.org/spatial#indexTask");
}
diff --git a/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/SpatialIndexFindUtils.java b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/SpatialIndexFindUtils.java
index 09120eba77..cc9efa7e42 100644
--- a/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/SpatialIndexFindUtils.java
+++ b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/SpatialIndexFindUtils.java
@@ -78,13 +78,13 @@ public class SpatialIndexFindUtils {
IteratorCloseable<SpatialIndexItem> result;
// Only add one set of statements as a converted dataset will duplicate the same info.
if (graph.contains(null, Geo.HAS_GEOMETRY_NODE, null)) {
- LOGGER.info("Feature-hasGeometry-Geometry statements found.");
- if (graph.contains(null, SpatialExtension.GEO_LAT_NODE, null)) {
- LOGGER.warn("Lat/Lon Geo predicates also found but will not be added to index.");
- }
+ // LOGGER.info("Feature-hasGeometry-Geometry statements found.");
+ // if (graph.contains(null, SpatialExtension.GEO_LAT_NODE, null)) {
+ // LOGGER.warn("Lat/Lon Geo predicates also found but will not be added to index.");
+ // }
result = findGeometryIndexItems(graph, srsURI);
} else if (graph.contains(null, SpatialExtension.GEO_LAT_NODE, null)) {
- LOGGER.info("Geo predicate statements found.");
+ // LOGGER.info("Geo predicate statements found.");
result = findGeoPredicateIndexItems(graph, srsURI);
} else {
result = Iter.empty();
@@ -92,6 +92,19 @@ public class SpatialIndexFindUtils {
return result;
}
+ /** Print out log messages for what type of spatial data is found in the given graph. */
+ public static final void checkSpatialIndexItems(Graph graph) {
+ // Only add one set of statements as a converted dataset will duplicate the same info.
+ if (graph.contains(null, Geo.HAS_GEOMETRY_NODE, null)) {
+ LOGGER.info("Feature-hasGeometry-Geometry statements found.");
+ if (graph.contains(null, SpatialExtension.GEO_LAT_NODE, null)) {
+ LOGGER.warn("Lat/Lon Geo predicates also found but will not be added to index.");
+ }
+ } else if (graph.contains(null, SpatialExtension.GEO_LAT_NODE, null)) {
+ LOGGER.info("Geo predicate statements found.");
+ }
+ }
+
/**
*
* @param graph
@@ -134,28 +147,35 @@ public class SpatialIndexFindUtils {
}
/**
- *
+ *
* @param graph
* @param srsURI
* @return Geo predicate objects prepared for adding to SpatialIndex.
*/
public static IteratorCloseable<SpatialIndexItem> findGeoPredicateIndexItems(Graph graph, String srsURI) {
+ // Warn about multiple lat/lon combinations only at most once per graph.
+ boolean enableWarnings = false;
+ boolean[] loggedMultipleLatLons = { false };
Iterator<Triple> latIt = graph.find(Node.ANY, SpatialExtension.GEO_LAT_NODE, Node.ANY);
IteratorCloseable<SpatialIndexItem> result = Iter.iter(latIt).flatMap(triple -> {
Node feature = triple.getSubject();
Node lat = triple.getObject();
- // Silently create the cross-product between lats and lons.
+ // Create the cross-product between lats and lons.
Iterator<Node> lons = G.iterSP(graph, feature, SpatialExtension.GEO_LON_NODE);
- // LOGGER.warn("Geo predicates: also found but will not be added to index.");
-
- // int[] lonCounter = {0};
+ // On malformed data this can cause lots of log output. Perhaps it's better to keep validation separate from indexing.
+ int[] lonCounter = {0};
Iterator<SpatialIndexItem> r = Iter.iter(lons).map(lon -> {
-// if (lonCounter[0] == 1) {
-// LOGGER.warn("Geo predicates: multiple longitudes on feature " + feature);
-// }
-// ++lonCounter[0];
+ if (enableWarnings) {
+ if (lonCounter[0] == 1) {
+ if (!loggedMultipleLatLons[0]) {
+ LOGGER.warn("Geo predicates: multiple longitudes detected on feature " + feature + ". Further warnings will be omitted.");
+ loggedMultipleLatLons[0] = true;
+ }
+ }
+ ++lonCounter[0];
+ }
GeometryWrapper geometryWrapper = ConvertLatLon.toGeometryWrapper(lat, lon);
SpatialIndexItem item = makeSpatialIndexItem(feature, geometryWrapper, srsURI);
return item;
@@ -166,7 +186,7 @@ public class SpatialIndexFindUtils {
}
public static SpatialIndexItem makeSpatialIndexItem(Node feature, GeometryWrapper geometryWrapper, String srsURI) {
- //Ensure all entries in the target SRS URI.
+ // Ensure all entries in the target SRS URI.
GeometryWrapper transformedGeometryWrapper = unsafeConvert(geometryWrapper, srsURI);
Envelope envelope = transformedGeometryWrapper.getEnvelope();
SpatialIndexItem item = new SpatialIndexItem(envelope, feature);
diff --git a/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v1/SpatialIndexV1.java b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v1/SpatialIndexV1.java
index cc89c70fea..5e5b27b655 100644
--- a/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v1/SpatialIndexV1.java
+++ b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v1/SpatialIndexV1.java
@@ -42,10 +42,11 @@ import org.apache.jena.geosparql.implementation.vocabulary.SRS_URI;
import org.apache.jena.geosparql.implementation.vocabulary.SpatialExtension;
import org.apache.jena.geosparql.spatial.ConvertLatLon;
import org.apache.jena.geosparql.spatial.SpatialIndex;
+import org.apache.jena.geosparql.spatial.SpatialIndexConstants;
import org.apache.jena.geosparql.spatial.SpatialIndexException;
import org.apache.jena.geosparql.spatial.SpatialIndexItem;
import org.apache.jena.geosparql.spatial.SpatialIndexStorage;
-import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexUtils;
+import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexLib;
import org.apache.jena.query.Dataset;
import org.apache.jena.query.DatasetFactory;
import org.apache.jena.query.ReadWrite;
@@ -218,7 +219,7 @@ public class SpatialIndexV1 {
public static final SpatialIndexV1 retrieve(ExecutionContext execCxt) throws SpatialIndexException {
Context context = execCxt.getContext();
- SpatialIndexV1 spatialIndex = (SpatialIndexV1) context.get(SpatialIndexUtils.SPATIAL_INDEX_SYMBOL, null);
+ SpatialIndexV1 spatialIndex = (SpatialIndexV1) context.get(SpatialIndexConstants.symSpatialIndex, null);
if (spatialIndex == null) {
throw new SpatialIndexException("Dataset Context does not contain SpatialIndex.");
@@ -234,7 +235,7 @@ public class SpatialIndexV1 {
*/
public static final boolean isDefined(ExecutionContext execCxt) {
Context context = execCxt.getContext();
- return context.isDefined(SpatialIndexUtils.SPATIAL_INDEX_SYMBOL);
+ return context.isDefined(SpatialIndexConstants.symSpatialIndex);
}
/**
@@ -247,7 +248,7 @@ public class SpatialIndexV1 {
public static final void setSpatialIndex(Dataset dataset, SpatialIndexV1 spatialIndex) {
Context context = dataset.getContext();
SpatialIndex wrapper = new SpatialIndexAdapterV1(spatialIndex);
- SpatialIndexUtils.setSpatialIndex(context, wrapper);
+ SpatialIndexLib.setSpatialIndex(context, wrapper);
}
/**
diff --git a/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v2/KryoRegistratorSpatialIndexV2.java b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v2/KryoRegistratorSpatialIndexV2.java
index 8b86e36375..a11738f4af 100644
--- a/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v2/KryoRegistratorSpatialIndexV2.java
+++ b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v2/KryoRegistratorSpatialIndexV2.java
@@ -46,7 +46,7 @@ import com.esotericsoftware.kryo.serializers.MapSerializer;
/**
* The class is used to configure the kryo serialization
- * of the spatial index. Changes to the kyro configuration may
+ * of the spatial index. Changes to the kryo configuration may
* break loading of existing index files - so changes must be made
* with care.
*/
@@ -55,7 +55,7 @@ public class KryoRegistratorSpatialIndexV2 {
private final static Logger LOGGER = LoggerFactory.getLogger(KryoRegistratorSpatialIndexV2.class);
public static void registerClasses(Kryo kryo, Serializer<Geometry> geometrySerializer) {
- LOGGER.debug("Registering custom serializers for geometry types");
+ LOGGER.debug("Registering kryo serializers for spatial index v2.");
// Java
Serializer<?> mapSerializer = new MapSerializer();
@@ -75,8 +75,10 @@ public class KryoRegistratorSpatialIndexV2 {
kryo.register(STRtreePerGraph.class, new STRtreePerGraphSerializer());
// The index only stores envelopes and jena nodes.
- // Therefore geometry serializers are not needed.
- // registerGeometrySerializers(kryo, geometrySerializer);
+ // Therefore, geometry serializers should not be needed.
+ if (geometrySerializer != null) {
+ registerGeometrySerializers(kryo, geometrySerializer);
+ }
}
/**
@@ -85,7 +87,9 @@ public class KryoRegistratorSpatialIndexV2 {
*/
public static void registerTripleSerializer(Kryo kryo) {
kryo.register(Triple.class, new TripleSerializer());
- // kryo.register(Triple[].class); // Array variants need to be registered though no extra serializer needed.
+
+ // Array-of-triples serializer variant would have to be registered separately - but does not seem to be needed.
+ // kryo.register(Triple[].class);
}
public static void registerGeometrySerializers(Kryo kryo, Serializer<Geometry> geometrySerializer) {
diff --git a/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v2/STRtreePerGraph.java b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v2/STRtreePerGraph.java
index d6f65c6024..6d534bb9a0 100644
--- a/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v2/STRtreePerGraph.java
+++ b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v2/STRtreePerGraph.java
@@ -39,6 +39,8 @@ public class STRtreePerGraph {
private static final Logger LOGGER = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
/**
+ * Mapping of graph node to STRtree.
+ *
* @implNote
* The STRtree that corresponds to the default graph (referred to as 'default tree')
* uses the name {@link Quad#defaultGraphIRI}.
@@ -47,12 +49,9 @@ public class STRtreePerGraph {
*/
private Map<Node, STRtree> treeMap;
- // Unmodifiable view
+ /** Unmodifiable view of treeMap */
private Map<Node, STRtree> treeMapView;
- // XXX Make srsInfo part of this class?
- // this.srsInfo = SRSRegistry.getSRSInfo(SRS_URI.DEFAULT_WKT_CRS84);
-
private boolean isBuilt = false;
public STRtreePerGraph() {
diff --git a/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v2/STRtreePerGraphSerializer.java b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v2/STRtreePerGraphSerializer.java
index 93e7008bd4..38ba56c7ee 100644
--- a/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v2/STRtreePerGraphSerializer.java
+++ b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v2/STRtreePerGraphSerializer.java
@@ -32,15 +32,15 @@ public class STRtreePerGraphSerializer
{
@Override
public void write(Kryo kryo, Output output, STRtreePerGraph index) {
- kryo.writeClassAndObject(output, index.getInternalTreeMap());
output.writeBoolean(index.isBuilt());
+ kryo.writeClassAndObject(output, index.getInternalTreeMap());
}
@Override
public STRtreePerGraph read(Kryo kryo, Input input, Class<STRtreePerGraph> type) {
+ boolean isBuilt = input.readBoolean();
@SuppressWarnings("unchecked")
Map<Node, STRtree> treeMap = (Map<Node, STRtree>)kryo.readClassAndObject(input);
- boolean isBuilt = input.readBoolean();
STRtreePerGraph result = new STRtreePerGraph(treeMap);
if (isBuilt) {
result.build();
diff --git a/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v2/STRtreeUtils.java b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v2/STRtreeUtils.java
index 0d17dd90bb..37775eae53 100644
--- a/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v2/STRtreeUtils.java
+++ b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v2/STRtreeUtils.java
@@ -17,7 +17,6 @@
*/
package org.apache.jena.geosparql.spatial.index.v2;
-import java.lang.invoke.MethodHandles;
import java.util.Collection;
import java.util.Iterator;
import java.util.LinkedHashMap;
@@ -31,15 +30,11 @@ import org.apache.jena.geosparql.spatial.SpatialIndexFindUtils;
import org.apache.jena.geosparql.spatial.SpatialIndexItem;
import org.apache.jena.graph.Graph;
import org.apache.jena.graph.Node;
-import org.apache.jena.graph.compose.Union;
import org.apache.jena.sparql.core.DatasetGraph;
import org.apache.jena.sparql.core.Quad;
import org.locationtech.jts.index.strtree.STRtree;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
public class STRtreeUtils {
- private static final Logger LOGGER = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
static final int MINIMUM_CAPACITY = 2;
public static STRtree buildSpatialIndexTree(Graph graph, String srsURI) throws SpatialIndexException {
@@ -57,27 +52,22 @@ public class STRtreeUtils {
}
}
- public static STRtree buildSpatialIndexTreeUnion(DatasetGraph datasetGraph, String srsURI) throws SpatialIndexException {
- // create the union view of default graph and all named graphs
- Union unionAllGraph = new Union(datasetGraph.getDefaultGraph(), datasetGraph.getUnionGraph());
- STRtree tree = buildSpatialIndexTree(unionAllGraph, srsURI);
- return tree;
- }
-
+ // XXX This method overlaps function-wise with SpatialIndexerComputation. Consolidate?
public static STRtreePerGraph buildSpatialIndexTree(DatasetGraph datasetGraph, String srsURI) throws SpatialIndexException {
Map<Node, STRtree> treeMap = new LinkedHashMap<>();
- LOGGER.info("building spatial index for default graph ...");
+ // Process default graph.
+ // LOGGER.info("building spatial index for default graph ...");
Graph defaultGraph = datasetGraph.getDefaultGraph();
STRtree defaultGraphTree = buildSpatialIndexTree(defaultGraph, srsURI);
treeMap.put(Quad.defaultGraphIRI, defaultGraphTree);
- // Named graphs
+ // Process named graphs.
Iterator<Node> graphIter = datasetGraph.listGraphNodes();
try {
while (graphIter.hasNext()) {
Node graphNode = graphIter.next();
- LOGGER.info("building spatial index for graph {} ...", graphNode);
+ // LOGGER.info("building spatial index for graph {} ...", graphNode);
Graph namedGraph = datasetGraph.getGraph(graphNode);
treeMap.put(graphNode, buildSpatialIndexTree(namedGraph, srsURI));
}
@@ -111,7 +101,6 @@ public class STRtreeUtils {
* It's the caller's responsibility to close the iterator if needed.
*/
public static void addToTree(STRtree treeAcc, Iterator<SpatialIndexItem> it) throws SpatialIndexException {
- // List<SpatialIndexItem> items = Iter.toList(it);
it.forEachRemaining(item -> treeAcc.insert(item.getEnvelope(), item.getItem()));
}
}
diff --git a/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v2/SpatialIndexIoKryo.java b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v2/SpatialIndexIoKryo.java
index 3f92547842..834c1cc8a6 100644
--- a/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v2/SpatialIndexIoKryo.java
+++ b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v2/SpatialIndexIoKryo.java
@@ -27,10 +27,8 @@ import java.util.Objects;
import org.apache.jena.atlas.RuntimeIOException;
import org.apache.jena.atlas.io.IOX;
-import org.apache.jena.geosparql.configuration.GeoSPARQLOperations;
import org.apache.jena.geosparql.implementation.SRSInfo;
import org.apache.jena.geosparql.implementation.registry.SRSRegistry;
-import org.apache.jena.geosparql.implementation.vocabulary.SRS_URI;
import org.apache.jena.geosparql.kryo.GeometrySerializerJtsWkb;
import org.apache.jena.geosparql.spatial.SpatialIndex;
import org.apache.jena.geosparql.spatial.SpatialIndexException;
@@ -49,25 +47,53 @@ import com.google.gson.JsonObject;
public class SpatialIndexIoKryo {
private static final Logger LOGGER = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- public static SpatialIndex buildSpatialIndex(Dataset dataset,
+ // Flag whether additional serializers for storing geometries in the index should be registered.
+ // The current version of the index only stores the envelopes so this feature is not needed.
+ private static boolean enableGeometrySerde = false;
+
+ public static SpatialIndex loadOrBuildSpatialIndex(Dataset dataset, Path spatialIndexFile) throws SpatialIndexException {
+ SpatialIndex spatialIndex = loadOrBuildSpatialIndex(dataset, null, spatialIndexFile);
+ return spatialIndex;
+ }
+
+ private static boolean isNonEmptyFile(Path path) {
+ boolean result = false;
+ if (path != null && Files.exists(path)) {
+ try {
+ result = Files.size(path) > 0;
+ } catch (IOException e) {
+ throw IOX.exception(e);
+ }
+ }
+ return result;
+ }
+
+ public static SpatialIndex loadOrBuildSpatialIndex(Dataset dataset,
String srsURI,
Path spatialIndexFile) throws SpatialIndexException {
+ SpatialIndexPerGraph spatialIndex = null;
- SpatialIndexPerGraph spatialIndex = load(spatialIndexFile);
-
- if (spatialIndex.isEmpty()) {
- spatialIndex = SpatialIndexUtils.buildSpatialIndex(dataset.asDatasetGraph(), srsURI);
- save(spatialIndexFile, spatialIndex);
+ // If the spatial index file exists and has non-zero size then load it.
+ // Otherwise build one.
+ if (isNonEmptyFile(spatialIndexFile)) {
+ spatialIndex = load(spatialIndexFile);
+ SpatialIndexLib.setSpatialIndex(dataset, spatialIndex);
+ } else {
+ spatialIndex = buildSpatialIndex(dataset, srsURI, spatialIndexFile);
}
- spatialIndex.setLocation(spatialIndexFile);
- SpatialIndexUtils.setSpatialIndex(dataset, spatialIndex);
return spatialIndex;
}
- public static SpatialIndex buildSpatialIndex(Dataset dataset, Path spatialIndexFile) throws SpatialIndexException {
- String srsURI = GeoSPARQLOperations.findModeSRS(dataset);
- SpatialIndex spatialIndex = buildSpatialIndex(dataset, srsURI, spatialIndexFile);
+ public static SpatialIndexPerGraph buildSpatialIndex(Dataset dataset,
+ String srsURI,
+ Path spatialIndexFile) throws SpatialIndexException {
+ SpatialIndexPerGraph spatialIndex = SpatialIndexLib.buildSpatialIndex(dataset.asDatasetGraph(), srsURI);
+ if (spatialIndexFile != null) {
+ // Register the source file with the index.
+ spatialIndex.setLocation(spatialIndexFile);
+ save(spatialIndexFile, spatialIndex);
+ }
return spatialIndex;
}
@@ -79,26 +105,39 @@ public class SpatialIndexIoKryo {
* @throws SpatialIndexException
*/
public static final void save(Path spatialIndexFile, SpatialIndexPerGraph index) throws SpatialIndexException {
- Path absPath = spatialIndexFile.toAbsolutePath();
- if (spatialIndexFile != null) {
- LOGGER.info("Saving Spatial Index - Started: {}", absPath);
+ Path originalFile = spatialIndexFile.toAbsolutePath();
+ LOGGER.info("Saving Spatial Index - Started: " + originalFile);
- String filename = absPath.toString();
- Path file = Path.of(filename);
- Path tmpFile = IOX.uniqueDerivedPath(file, null);
- try {
- Files.deleteIfExists(file);
- } catch (IOException ex) {
- throw new SpatialIndexException("Failed to delete file: " + ex.getMessage());
- }
+ // Create a temporary file for writing the new index.
+ Path tmpFile = IOX.uniqueDerivedPath(originalFile, null);
+
+ // As long as the new file has not been successfully written:
+ // Move the original file out of the way but don't delete it yet.
+ Path originalBackup = IOX.uniqueDerivedPath(originalFile, baseName -> baseName + ".bak");
+ if (Files.exists(originalFile)) {
+ IOX.moveAllowCopy(originalFile, originalBackup);
+ }
+
+ try {
+ IOX.safeWriteOrCopy(originalFile, tmpFile, out -> writeToOutputStream(out, index));
+ LOGGER.info("Saving Spatial Index - Success: " + originalFile);
+ } catch (RuntimeIOException ex) {
+ LOGGER.info("Failure writing spatial index: " + originalFile, ex);
+ // Attempt to restore original file from backed up one.
try {
- IOX.safeWriteOrCopy(file, tmpFile, out -> writeToOutputStream(out, index));
- } catch (RuntimeIOException ex) {
- throw new SpatialIndexException("Save Exception: " + ex.getMessage());
- } finally {
- LOGGER.info("Saving Spatial Index - Completed: {}", absPath);
+ IOX.moveAllowCopy(originalBackup, originalFile);
+ } catch (RuntimeException ex2) {
+ LOGGER.warn("Failed to restore " + originalFile + " + from backup file " + originalBackup, ex2);
}
+ throw new SpatialIndexException("Save Exception: " + originalFile + " (via temp file: " + tmpFile + ")", ex);
+ }
+
+ // Delete backup
+ try {
+ Files.deleteIfExists(originalBackup);
+ } catch (IOException ex) {
+ LOGGER.warn("Failed to remove no longer needed backup: " + originalBackup, ex);
}
}
@@ -108,16 +147,19 @@ public class SpatialIndexIoKryo {
* @param index spatial index
*/
public static void writeToOutputStream(OutputStream os, SpatialIndexPerGraph index) {
- GeometrySerializerJtsWkb geometrySerde = new GeometrySerializerJtsWkb();
-
SpatialIndexHeader header = new SpatialIndexHeader();
header.setType(SpatialIndexHeader.TYPE_VALUE);
header.setVersion("2.0.0");
header.setSrsUri(index.getSrsInfo().getSrsURI());
- header.setGeometrySerializerClass(geometrySerde.getClass().getName());
+
+ GeometrySerializerJtsWkb geometrySerializer = null;
+ if (enableGeometrySerde) {
+ geometrySerializer = new GeometrySerializerJtsWkb();
+ header.setGeometrySerializerClass(geometrySerializer.getClass().getName());
+ }
Kryo kryo = new Kryo();
- KryoRegistratorSpatialIndexV2.registerClasses(kryo, geometrySerde);
+ KryoRegistratorSpatialIndexV2.registerClasses(kryo, geometrySerializer);
try (Output output = new Output(os)) {
writeHeader(output, header);
STRtreePerGraph trees = index.getIndex();
@@ -153,30 +195,29 @@ public class SpatialIndexIoKryo {
String srsUri;
STRtreePerGraph index;
- if (spatialIndexFile != null && Files.exists(spatialIndexFile)) {
- spatialIndexFile = spatialIndexFile.toAbsolutePath();
- LOGGER.info("Loading Spatial Index - Started: {}", spatialIndexFile);
+ spatialIndexFile = spatialIndexFile.toAbsolutePath();
+ LOGGER.info("Loading Spatial Index - Started: {}", spatialIndexFile);
- try (Input input = new Input(Files.newInputStream(spatialIndexFile))) {
- SpatialIndexHeader header = readHeader(input);
+ try (Input input = new Input(Files.newInputStream(spatialIndexFile))) {
+ SpatialIndexHeader header = readHeader(input);
- String type = header.getType();
- if (!"jena-spatial-index".equals(type)) {
- throw new RuntimeException("Type does not indicate a spatial index file.");
- }
+ String type = header.getType();
+ if (!"jena-spatial-index".equals(type)) {
+ throw new RuntimeException("Type does not indicate a spatial index file.");
+ }
- String version = header.getVersion();
- if (!"2.0.0".equals(version)) {
- throw new SpatialIndexException("The version of the spatial index does not match the version of this loader class.");
- }
+ String version = header.getVersion();
+ if (!"2.0.0".equals(version)) {
+ throw new SpatialIndexException("The version of the spatial index does not match the version of this loader class.");
+ }
- srsUri = header.getSrsUri();
+ srsUri = header.getSrsUri();
- // Get the geometrySerializer attribute
+ Serializer<Geometry> geometrySerializer = null;
+ if (enableGeometrySerde) {
String geometrySerdeName = header.getGeometrySerializerClass();
Objects.requireNonNull(geometrySerdeName, "Field 'geometrySerde' not set.");
- Serializer<Geometry> geometrySerializer;
try {
Class<?> geometrySerdeClass = Class.forName(geometrySerdeName);
geometrySerializer = (Serializer<Geometry>)geometrySerdeClass.getConstructor().newInstance();
@@ -184,19 +225,15 @@ public class SpatialIndexIoKryo {
| InvocationTargetException | NoSuchMethodException | ClassCastException | SecurityException e) {
throw new SpatialIndexException("Failed to load index", e);
}
+ }
- Kryo kryo = new Kryo();
- KryoRegistratorSpatialIndexV2.registerClasses(kryo, geometrySerializer);
+ Kryo kryo = new Kryo();
+ KryoRegistratorSpatialIndexV2.registerClasses(kryo, geometrySerializer);
- index = kryo.readObject(input, STRtreePerGraph.class);
- LOGGER.info("Loading Spatial Index - Completed: {}", spatialIndexFile);
- } catch (IOException ex) {
- throw new SpatialIndexException("Loading Exception: " + ex.getMessage(), ex);
- }
- } else {
- LOGGER.info("File {} does not exist. Creating empty Spatial Index.", (spatialIndexFile != null ? spatialIndexFile.toAbsolutePath() : "null"));
- srsUri = SRS_URI.DEFAULT_WKT_CRS84;
- index = new STRtreePerGraph();
+ index = kryo.readObject(input, STRtreePerGraph.class);
+ LOGGER.info("Loading Spatial Index - Completed: {}", spatialIndexFile);
+ } catch (IOException ex) {
+ throw new SpatialIndexException("Loading Exception: " + ex.getMessage(), ex);
}
SRSInfo srsInfo = SRSRegistry.getSRSInfo(srsUri);
diff --git a/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v2/SpatialIndexLib.java b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v2/SpatialIndexLib.java
new file mode 100644
index 0000000000..bcb35c4740
--- /dev/null
+++ b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v2/SpatialIndexLib.java
@@ -0,0 +1,409 @@
+/*
+ * 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.jena.geosparql.spatial.index.v2;
+
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.Predicate;
+import java.util.stream.Stream;
+
+import org.apache.jena.atlas.iterator.Iter;
+import org.apache.jena.geosparql.configuration.GeoSPARQLOperations;
+import org.apache.jena.geosparql.implementation.SRSInfo;
+import org.apache.jena.geosparql.implementation.registry.SRSRegistry;
+import org.apache.jena.geosparql.spatial.SpatialIndex;
+import org.apache.jena.geosparql.spatial.SpatialIndexConstants;
+import org.apache.jena.geosparql.spatial.SpatialIndexException;
+import org.apache.jena.geosparql.spatial.task.TaskThread;
+import org.apache.jena.geosparql.spatial.task.BasicTask;
+import org.apache.jena.geosparql.spatial.task.BasicTask.TaskListener;
+import org.apache.jena.graph.Graph;
+import org.apache.jena.graph.Node;
+import org.apache.jena.query.Dataset;
+import org.apache.jena.query.DatasetFactory;
+import org.apache.jena.query.TxnType;
+import org.apache.jena.rdf.model.Model;
+import org.apache.jena.sparql.core.DatasetGraph;
+import org.apache.jena.sparql.core.NamedGraph;
+import org.apache.jena.sparql.core.Quad;
+import org.apache.jena.sparql.engine.ExecutionContext;
+import org.apache.jena.sparql.util.Context;
+import org.apache.jena.system.AutoTxn;
+import org.apache.jena.system.Txn;
+import org.locationtech.jts.index.strtree.STRtree;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** ARQ-level utils for Dataset and Context. */
+public class SpatialIndexLib {
+
+ private static final Logger logger = LoggerFactory.getLogger(SpatialIndexLib.class);
+
+ /** Set the preferred SRS for the given (dataset) context. */
+ public static final String getPreferredSRS(Context context) {
+ return context == null ? null : context.getAsString(SpatialIndexConstants.symSrsUri);
+ }
+
+ /** Set a preferred SRS for the given (dataset) context. This SRS will be used when constructing a spatial index for the first time. */
+ public static final void setPreferredSRS(Context context, String srsUri) {
+ context.set(SpatialIndexConstants.symSrsUri, srsUri);
+ }
+
+ /**
+ * Set the SpatialIndex into the Context of the Dataset for later retrieval
+ * and use in spatial functions.
+ *
+ * @param dataset
+ * @param spatialIndex
+ */
+ public static final void setSpatialIndex(Dataset dataset, SpatialIndex spatialIndex) {
+ Context cxt = dataset.getContext();
+ setSpatialIndex(cxt, spatialIndex);
+ }
+
+ public static final void setSpatialIndex(DatasetGraph datasetGraph, SpatialIndex spatialIndex) {
+ Context cxt = datasetGraph.getContext();
+ setSpatialIndex(cxt, spatialIndex);
+ }
+
+ public static final void setSpatialIndex(Context context, SpatialIndex spatialIndex) {
+ context.set(SpatialIndexConstants.symSpatialIndex, spatialIndex);
+ }
+
+ public static final <T extends SpatialIndex> T getSpatialIndex(Context cxt) {
+ return cxt == null ? null : cxt.get(SpatialIndexConstants.symSpatialIndex);
+ }
+
+ public static final <T extends SpatialIndex> T getSpatialIndex(DatasetGraph dsg) {
+ return getSpatialIndex(dsg.getContext());
+ }
+
+ /**
+ *
+ * @param execCxt
+ * @return True if a SpatialIndex is defined in the ExecutionContext.
+ */
+ public static final boolean isDefined(ExecutionContext execCxt) {
+ Context context = execCxt.getContext();
+ return context.isDefined(SpatialIndexConstants.symSpatialIndex);
+ }
+
+ /**
+ * Retrieve the SpatialIndex from the Context.
+ *
+ * @param execCxt
+ * @return SpatialIndex contained in the Context.
+ * @throws SpatialIndexException
+ */
+ public static final SpatialIndex retrieve(ExecutionContext execCxt) throws SpatialIndexException {
+ Context context = execCxt.getContext();
+ SpatialIndex spatialIndex = (SpatialIndex) context.get(SpatialIndexConstants.symSpatialIndex, null);
+ if (spatialIndex == null) {
+ throw new SpatialIndexException("Dataset Context does not contain SpatialIndex.");
+ }
+ return spatialIndex;
+ }
+
+ /**
+ * Wrap Model in a Dataset and build SpatialIndex.
+ *
+ * @param model
+ * @param srsURI
+ * @return Dataset with default Model and SpatialIndex in Context.
+ * @throws SpatialIndexException
+ */
+ public static final Dataset wrapModel(Model model, String srsURI) throws SpatialIndexException {
+ Dataset dataset = DatasetFactory.createTxnMem();
+ dataset.setDefaultModel(model);
+ buildSpatialIndex(dataset.asDatasetGraph(), srsURI);
+ return dataset;
+ }
+
+ /**
+ * Wrap Model in a Dataset and build SpatialIndex.
+ *
+ * @param model
+ * @return Dataset with default Model and SpatialIndex in Context.
+ * @throws SpatialIndexException
+ */
+ public static final Dataset wrapModel(Model model) throws SpatialIndexException {
+ Dataset dataset = DatasetFactory.createTxnMem();
+ dataset.setDefaultModel(model);
+ String srsURI = GeoSPARQLOperations.findModeSRS(dataset);
+ buildSpatialIndex(dataset.asDatasetGraph(), srsURI);
+ return dataset;
+ }
+
+ /**
+ * Build Spatial Index from all graphs in Dataset.<br>
+ * Dataset contains SpatialIndex in Context.<br>
+ * SRS URI based on most frequent found in Dataset.
+ *
+ * @param datasetGraph
+ * @return SpatialIndex constructed.
+ * @throws SpatialIndexException
+ */
+ public static SpatialIndex buildSpatialIndex(DatasetGraph datasetGraph) throws SpatialIndexException {
+ SpatialIndex spatialIndex = buildSpatialIndex(datasetGraph, null);
+ return spatialIndex;
+ }
+
+ /**
+ * Build Spatial Index from all graphs in Dataset.<br>
+ * Dataset contains SpatialIndex in Context.
+ *
+ * @param datasetGraph
+ * @param srsURI
+ * @return SpatialIndex constructed.
+ * @throws SpatialIndexException
+ */
+ public static SpatialIndexPerGraph buildSpatialIndex(DatasetGraph datasetGraph, String srsURI) throws SpatialIndexException {
+ return buildSpatialIndexPerGraph(datasetGraph, srsURI);
+ }
+
+ public static SpatialIndexPerGraph buildSpatialIndexPerGraph(DatasetGraph datasetGraph, String srsURI) throws SpatialIndexException {
+ Objects.requireNonNull(datasetGraph);
+
+ if (srsURI == null) {
+ // XXX Dataset wrapping due to legacy code.
+ Dataset dataset = DatasetFactory.wrap(datasetGraph);
+ srsURI = GeoSPARQLOperations.findModeSRS(dataset);
+ }
+
+ // XXX SpatialIndexerComputation could be adapted to run the code below if just 1 thread is requested.
+ // SpatialIndexerComputation computation = new SpatialIndexerComputation(datasetGraph, srsURI, null, 1);
+
+ STRtreePerGraph treePerGraph;
+ logger.info("Building Spatial Index - Started");
+ try (AutoTxn txn = Txn.autoTxn(datasetGraph, TxnType.READ)) {
+ treePerGraph = STRtreeUtils.buildSpatialIndexTree(datasetGraph, srsURI);
+ txn.commit();
+ }
+ logger.info("Building Spatial Index - Completed");
+
+ SRSInfo srsInfo = SRSRegistry.getSRSInfo(srsURI);
+ SpatialIndexPerGraph index = new SpatialIndexPerGraph(srsInfo, treePerGraph, null);
+ setSpatialIndex(datasetGraph, index);
+ return index;
+ }
+
+ public static Node unwrapGraphName(Graph graph) {
+ Node graphNode = graph instanceof NamedGraph namedGraph
+ ? namedGraph.getGraphName()
+ : null;
+ return graphNode;
+ }
+
+ public static BasicTask scheduleOnceIndexTask(DatasetGraph dsg, SpatialIndexerComputation indexComputation, Path targetFile, boolean isReplaceTask,
+ TaskListener<BasicTask> taskListener) {
+ Context cxt = dsg.getContext();
+
+ BasicTask task = cxt.compute(SpatialIndexConstants.symSpatialIndexTask, (key, priorTaskObj) -> {
+ BasicTask priorTask = (BasicTask)priorTaskObj;
+ if (priorTask != null && !priorTask.isTerminated()) {
+ throw new RuntimeException("A spatial indexing task is already active for this dataset. Wait for completion or abort it.");
+ }
+
+ TaskThread thread = createIndexerTask(dsg, null, indexComputation, taskListener, targetFile, isReplaceTask);
+ thread.start();
+ return thread;
+ });
+
+ return task;
+ }
+
+ public static TaskThread createIndexerTask(DatasetGraph dsg, Predicate<Node> isAuthorizedGraph, SpatialIndexerComputation indexComputation, TaskListener<BasicTask> taskListener, Path targetFile, boolean isReplaceTask) {
+ Context cxt = dsg.getContext();
+ long graphCount = indexComputation.getGraphNodes().size();
+ boolean isEffectiveUpdate = !isReplaceTask || isAuthorizedGraph != null;
+
+ TaskThread thread = new TaskThread("Spatial Indexer Task", taskListener) {
+ @Override
+ public void runActual() throws Exception {
+ // Prevent deletions of graphs from the index which a user cannot see due to access restrictions.
+ // For this we need the physical graph list.
+ // Replace task on a visible subset is an update.
+ // With removal of all graphs in visible but not in selected.
+
+ if (logger.isInfoEnabled()) {
+ String replaceMsg = isReplaceTask ? "The resulting index will REPLACE a prior index." : "A prior index will be UPDATED with the newly indexed graphs.";
+ logger.info("Indexing of {} graphs started. " + replaceMsg, graphCount);
+ }
+
+ SpatialIndexPerGraph rawOldIndex = SpatialIndexLib.getSpatialIndex(cxt);
+
+ // Check that prior and new SRS are consistent.
+ if (isEffectiveUpdate) {
+ String priorSrs = Optional.ofNullable(rawOldIndex)
+ .map(SpatialIndex::getSrsInfo).map(SRSInfo::getSrsURI).orElse(null);
+ String requestedSrs = indexComputation.getSrsURI();
+
+ if (priorSrs != null && !priorSrs.equals(requestedSrs)) {
+ throw new IllegalArgumentException("The SRS of the update request is inconistent with the SRS of the index: index SRS: " + priorSrs + ", requested SRS: " + requestedSrs);
+ }
+ }
+
+ // Uncomment to test artificial delays.
+ // Thread.sleep(5000);
+
+ SpatialIndexPerGraph newIndex = indexComputation.call();
+
+ // If NOT in replace mode, add all graph-indexes from the previous index
+ if (isEffectiveUpdate) {
+ // Copy the old index into a new one.
+ SpatialIndexPerGraph oldIndex = (SpatialIndexPerGraph)rawOldIndex;
+ if (oldIndex != null) {
+ Map<Node, STRtree> oldTreeMap = oldIndex.getIndex().getTreeMap();
+
+ oldTreeMap.forEach((name, tree) -> {
+ boolean isGraphNotSelectedForUpdate = !newIndex.getIndex().contains(name);
+
+ boolean addGraph = false;
+ if (isReplaceTask) {
+ boolean isNonAuthorizedGraph = isAuthorizedGraph != null && !isAuthorizedGraph.test(name);
+ if (isNonAuthorizedGraph) {
+ addGraph = true;
+ }
+ } else {
+ addGraph = isGraphNotSelectedForUpdate;
+ }
+
+ if (addGraph) {
+ newIndex.getIndex().setTree(name, tree);
+ }
+ });
+ }
+ }
+
+ SpatialIndexLib.setSpatialIndex(cxt, newIndex);
+ if (targetFile != null) {
+ newIndex.setLocation(targetFile);
+ logger.info("Writing spatial index of {} graphs to disk at path {}", graphCount, targetFile.toAbsolutePath());
+ SpatialIndexIoKryo.save(targetFile, newIndex);
+ }
+ String statusMsg = String.format("Updated spatial index with %d graphs.", graphCount);
+ setStatusMessage(statusMsg);
+ if (logger.isInfoEnabled()) {
+ logger.info("Indexing of {} graphs completed successfully.", graphCount);
+ }
+ }
+
+ @Override
+ public void requestCancel() {
+ indexComputation.abort();
+ super.requestCancel(); // Interrupt
+ }
+ };
+
+ return thread;
+ }
+
+ /**
+ * Attempt to start a spatial index task that cleans the index of graphs not present in the given dataset.
+ * This method fails if there is already another spatial index task running.
+ */
+ public static BasicTask scheduleOnceCleanTask(DatasetGraph dsg, TaskListener<BasicTask> taskListener) {
+ Context cxt = dsg.getContext();
+ BasicTask task = cxt.compute(SpatialIndexConstants.symSpatialIndexTask, (key, priorTaskObj) -> {
+ BasicTask priorTask = (BasicTask) priorTaskObj;
+ if (priorTask != null && !priorTask.isTerminated()) {
+ throw new RuntimeException("A spatial indexing task is already active for this dataset. Wait for completion or abort it.");
+ }
+
+ TaskThread thread = createCleanTask(dsg, null, taskListener);
+ thread.start();
+ return thread;
+ });
+ return task;
+ }
+
+ public static TaskThread createCleanTask(DatasetGraph dsg, Predicate<Node> isAuthorizedGraph, TaskListener<BasicTask> taskListener) {
+ Context cxt = dsg.getContext();
+
+ TaskThread thread = new TaskThread("Clean action", taskListener) {
+ @Override
+ public void runActual() throws Exception {
+ SpatialIndex spatialIndexRaw = SpatialIndexLib.getSpatialIndex(cxt);
+ if (spatialIndexRaw == null) {
+ throw new SpatialIndexException("No spatial index available on current dataset.");
+ } else if (spatialIndexRaw instanceof SpatialIndexPerGraph spatialIndex) {
+ // Prevent deletions of graphs from the index which a user cannot see due to access restrictions.
+ // For this we need the physical graph list.
+ Set<Node> physicalGraphs = Txn.calculateRead(dsg, () -> accGraphNodes(new LinkedHashSet<>(), dsg));
+
+ STRtreePerGraph perGraphIndex = spatialIndex.getIndex();
+ Map<Node, STRtree> treeMap = perGraphIndex.getTreeMap();
+ Set<Node> visibleGraphNodes = Txn.calculateRead(dsg, () -> accGraphNodes(new LinkedHashSet<>(), dsg));
+ if (physicalGraphs == null) {
+ physicalGraphs = visibleGraphNodes;
+ visibleGraphNodes = null;
+ }
+
+ List<Node> indexGraphNodes = new ArrayList<>(treeMap.keySet());
+
+ int cleanCount = 0;
+ for (Node node : indexGraphNodes) {
+ if (!(node == null || Quad.isDefaultGraph(node))) { // Can never delete the default graph.
+ // A graph is only subject to removal if it is not in the physical graph list.
+ if (!physicalGraphs.contains(node)) {
+ // If the graph subject to removal is not visible to the user then reject the removal
+ if (isAuthorizedGraph != null && !isAuthorizedGraph.test(node)) {
+ // Prevent removal of a graph we are not authorized to.
+ continue;
+ }
+ perGraphIndex.removeTree(node);
+ ++cleanCount;
+ // System.out.println("Removed: [" + node + "] " + (node == null));
+ }
+ }
+ }
+
+ int finalGraphCount = treeMap.keySet().size();
+ Path targetFile = spatialIndex.getLocation();
+ // SpatialIndexUtils.setSpatialIndex(cxt, newIndex);
+ if (cleanCount > 0 && targetFile != null) {
+ // newIndex.setLocation(targetFile);
+ logger.info("Writing spatial index of {} graphs (cleaned: {}) to disk at path {}", finalGraphCount, cleanCount, targetFile.toAbsolutePath());
+ SpatialIndexIoKryo.save(targetFile, spatialIndex);
+ }
+ String statusMsg = String.format("Updated spatial index of %d graphs (cleaned: %d)", finalGraphCount, cleanCount);
+ setStatusMessage(statusMsg);
+ logger.info("Indexing of {} graphs completed successfully.", finalGraphCount);
+ } else {
+ throw new SpatialIndexException("Unsupported spatial index type for cleaning.");
+ }
+ }
+ };
+ return thread;
+ }
+
+ public static <C extends Collection<Node>> C accGraphNodes(C accGraphs, DatasetGraph dsg) {
+ try (Stream<Node> s = Iter.asStream(dsg.listGraphNodes())) {
+ s.forEach(accGraphs::add);
+ }
+ return accGraphs;
+ }
+}
diff --git a/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v2/SpatialIndexUtils.java b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v2/SpatialIndexUtils.java
deleted file mode 100644
index 1d055a4e09..0000000000
--- a/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v2/SpatialIndexUtils.java
+++ /dev/null
@@ -1,215 +0,0 @@
-/*
- * 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.jena.geosparql.spatial.index.v2;
-
-import java.lang.invoke.MethodHandles;
-
-import org.apache.jena.geosparql.configuration.GeoSPARQLOperations;
-import org.apache.jena.geosparql.implementation.SRSInfo;
-import org.apache.jena.geosparql.implementation.registry.SRSRegistry;
-import org.apache.jena.geosparql.spatial.SpatialIndex;
-import org.apache.jena.geosparql.spatial.SpatialIndexException;
-import org.apache.jena.graph.Graph;
-import org.apache.jena.graph.Node;
-import org.apache.jena.query.Dataset;
-import org.apache.jena.query.DatasetFactory;
-import org.apache.jena.query.TxnType;
-import org.apache.jena.rdf.model.Model;
-import org.apache.jena.sparql.core.DatasetGraph;
-import org.apache.jena.sparql.core.NamedGraph;
-import org.apache.jena.sparql.engine.ExecutionContext;
-import org.apache.jena.sparql.util.Context;
-import org.apache.jena.sparql.util.Symbol;
-import org.apache.jena.system.AutoTxn;
-import org.apache.jena.system.Txn;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/** ARQ-level utils for Dataset and Context. */
-public class SpatialIndexUtils {
- private static final Logger LOGGER = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
-
- public static final Symbol SPATIAL_INDEX_SYMBOL = Symbol.create("http://jena.apache.org/spatial#index");
-
- /** Symbol for a running task in a dataset's context. */
- public static final Symbol SPATIAL_INDEX_TASK_SYMBOL = Symbol.create("http://jena.apache.org/spatial#indexTask");
- public static final Symbol symSrsUri = Symbol.create("http://jena.apache.org/spatial#srsURI");
-
- /**
- * Set the SpatialIndex into the Context of the Dataset for later retrieval
- * and use in spatial functions.
- *
- * @param dataset
- * @param spatialIndex
- */
- public static final void setSpatialIndex(Dataset dataset, SpatialIndex spatialIndex) {
- Context cxt = dataset.getContext();
- setSpatialIndex(cxt, spatialIndex);
- }
-
- public static final void setSpatialIndex(DatasetGraph datasetGraph, SpatialIndex spatialIndex) {
- Context cxt = datasetGraph.getContext();
- setSpatialIndex(cxt, spatialIndex);
- }
-
- public static final void setSpatialIndex(Context context, SpatialIndex spatialIndex) {
- context.set(SPATIAL_INDEX_SYMBOL, spatialIndex);
- }
-
- public static final SpatialIndex getSpatialIndex(Context cxt) {
- return cxt.get(SPATIAL_INDEX_SYMBOL);
- }
-
- /**
- *
- * @param execCxt
- * @return True if a SpatialIndex is defined in the ExecutionContext.
- */
- public static final boolean isDefined(ExecutionContext execCxt) {
- Context context = execCxt.getContext();
- return context.isDefined(SPATIAL_INDEX_SYMBOL);
- }
-
- /**
- * Retrieve the SpatialIndex from the Context.
- *
- * @param execCxt
- * @return SpatialIndex contained in the Context.
- * @throws SpatialIndexException
- */
- public static final SpatialIndex retrieve(ExecutionContext execCxt) throws SpatialIndexException {
- Context context = execCxt.getContext();
- SpatialIndex spatialIndex = (SpatialIndex) context.get(SPATIAL_INDEX_SYMBOL, null);
- if (spatialIndex == null) {
- throw new SpatialIndexException("Dataset Context does not contain SpatialIndex.");
- }
- return spatialIndex;
- }
-
- /**
- * Wrap Model in a Dataset and build SpatialIndex.
- *
- * @param model
- * @param srsURI
- * @return Dataset with default Model and SpatialIndex in Context.
- * @throws SpatialIndexException
- */
- public static final Dataset wrapModel(Model model, String srsURI) throws SpatialIndexException {
- Dataset dataset = DatasetFactory.createTxnMem();
- dataset.setDefaultModel(model);
- buildSpatialIndex(dataset.asDatasetGraph(), srsURI);
- return dataset;
- }
-
- /**
- * Wrap Model in a Dataset and build SpatialIndex.
- *
- * @param model
- * @return Dataset with default Model and SpatialIndex in Context.
- * @throws SpatialIndexException
- */
- public static final Dataset wrapModel(Model model) throws SpatialIndexException {
- Dataset dataset = DatasetFactory.createTxnMem();
- dataset.setDefaultModel(model);
- String srsURI = GeoSPARQLOperations.findModeSRS(dataset);
- buildSpatialIndex(dataset.asDatasetGraph(), srsURI);
- return dataset;
- }
-
- /**
- * Build Spatial Index from all graphs in Dataset.<br>
- * Dataset contains SpatialIndex in Context.<br>
- * SRS URI based on most frequent found in Dataset.
- *
- * @param datasetGraph
- * @return SpatialIndex constructed.
- * @throws SpatialIndexException
- */
- public static SpatialIndex buildSpatialIndex(DatasetGraph datasetGraph) throws SpatialIndexException {
- // XXX Dataset wrapping due to legacy code
- Dataset dataset = DatasetFactory.wrap(datasetGraph);
- String srsURI = GeoSPARQLOperations.findModeSRS(dataset);
- SpatialIndex spatialIndex = buildSpatialIndex(datasetGraph, srsURI);
- return spatialIndex;
- }
-
- /**
- * Build Spatial Index from all graphs in Dataset.<br>
- * Dataset contains SpatialIndex in Context.
- *
- * @param datasetGraph
- * @param srsURI
- * @return SpatialIndex constructed.
- * @throws SpatialIndexException
- */
- public static SpatialIndexPerGraph buildSpatialIndex(DatasetGraph datasetGraph, String srsURI) throws SpatialIndexException {
- return buildSpatialIndexPerGraph(datasetGraph, srsURI);
- }
-
-// public static SpatialIndexPerGraph buildSpatialIndex(DatasetGraph datasetGraph, String srsURI, boolean indexTreePerGraph) throws SpatialIndexException {
-// SpatialIndexPerGraph result = indexTreePerGraph
-// ? buildSpatialIndexPerGraph(datasetGraph, srsURI)
-// : buildSpatialIndexUnion(datasetGraph, srsURI);
-// return result;
-// }
-
-// public static SpatialIndexPerGraph buildSpatialIndexUnion(DatasetGraph datasetGraph, String srsURI) throws SpatialIndexException {
-// // we always compute an index tree DGT for the default graph
-// // if an index per named graph NG is enabled, we compute a separate index tree NGT for each NG, otherwise all
-// // items will be indexed in the default graph index tree DGT
-//
-// STRtree treePerGraph;
-// LOGGER.info("Building Spatial Index - Started");
-// try (AutoTxn txn = Txn.begin(datasetGraph, TxnType.READ)) {
-// treePerGraph = STRtreeUtils.buildSpatialIndexTreeUnion(datasetGraph, srsURI);
-// }
-// LOGGER.info("Building Spatial Index - Completed");
-//
-// SRSInfo srsInfo = SRSRegistry.getSRSInfo(srsURI);
-// STRtreePerGraph trees = new STRtreePerGraph(treePerGraph);
-// SpatialIndexPerGraph index = new SpatialIndexPerGraph(srsInfo, trees, null);
-// setSpatialIndex(datasetGraph, index);
-// return index;
-// }
-
- public static SpatialIndexPerGraph buildSpatialIndexPerGraph(DatasetGraph datasetGraph, String srsURI) throws SpatialIndexException {
- // we always compute an index tree DGT for the default graph
- // if an index per named graph NG is enabled, we compute a separate index tree NGT for each NG, otherwise all
- // items will be indexed in the default graph index tree DGT
-
- STRtreePerGraph treePerGraph;
- LOGGER.info("Building Spatial Index - Started");
- try (AutoTxn txn = Txn.autoTxn(datasetGraph, TxnType.READ)) {
- treePerGraph = STRtreeUtils.buildSpatialIndexTree(datasetGraph, srsURI);
- txn.commit();
- }
- LOGGER.info("Building Spatial Index - Completed");
-
- SRSInfo srsInfo = SRSRegistry.getSRSInfo(srsURI);
- SpatialIndexPerGraph index = new SpatialIndexPerGraph(srsInfo, treePerGraph, null);
- setSpatialIndex(datasetGraph, index);
- return index;
- }
-
- public static Node unwrapGraphName(Graph graph) {
- Node graphNode = graph instanceof NamedGraph namedGraph
- ? namedGraph.getGraphName()
- : null;
- return graphNode;
- }
-}
diff --git a/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v2/SpatialIndexerComputation.java b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v2/SpatialIndexerComputation.java
index eb7c182b53..ada66757c6 100644
--- a/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v2/SpatialIndexerComputation.java
+++ b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/index/v2/SpatialIndexerComputation.java
@@ -24,6 +24,7 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
+import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
@@ -38,13 +39,18 @@ import org.apache.jena.graph.Node;
import org.apache.jena.query.TxnType;
import org.apache.jena.sparql.core.DatasetGraph;
import org.apache.jena.sparql.core.Quad;
+import org.apache.jena.sparql.engine.iterator.Abortable;
import org.apache.jena.system.AutoTxn;
import org.apache.jena.system.Txn;
import org.locationtech.jts.index.strtree.STRtree;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class SpatialIndexerComputation {
+
+/** Low level class to compute geo indexes for a given set of graphs. */
+public class SpatialIndexerComputation
+ implements Callable<SpatialIndexPerGraph>, Abortable
+{
private static final Logger LOGGER = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
private DatasetGraph datasetGraph;
@@ -59,6 +65,8 @@ public class SpatialIndexerComputation {
private List<Future<Entry<Node, STRtree>>> futures = new ArrayList<>();
+ private static boolean logProgress = false;
+
/**
* Create an instance of a spatial indexer computation.
*
@@ -78,31 +86,52 @@ public class SpatialIndexerComputation {
}
}
+ public String getSrsURI() {
+ return srsURI;
+ }
+
public List<Node> getGraphNodes() {
return graphNodes;
}
private Entry<Node, STRtree> indexOneGraph(Node graphNode) throws SpatialIndexException {
- LOGGER.info("Started spatial index build for graph {} ...", graphNode);
- STRtree tree;
+ if (logProgress) {
+ LOGGER.info("Started spatial index build for graph {} ...", graphNode);
+ }
+
+ STRtree tree = null;
try (AutoTxn txn = Txn.autoTxn(datasetGraph, TxnType.READ)) {
Graph graph = Quad.isDefaultGraph(graphNode) // XXX would getGraph work with the default graph URIs?
? datasetGraph.getDefaultGraph()
: datasetGraph.getGraph(graphNode);
- tree = STRtreeUtils.buildSpatialIndexTree(graph, srsURI);
+ if (graph != null) { // May be null if the requested graph does not exist (possibly due to a dynamic dataset)
+ tree = STRtreeUtils.buildSpatialIndexTree(graph, srsURI);
+ }
// XXX This commit is a workaround for DatasetGraphText.abort() causing a NPE in
// during multi-threaded spatial index computation.
// This is an issue related to the transaction mechanics of DatasetGraphText.
txn.commit();
}
- LOGGER.info("Completed spatial index for graph {}", graphNode);
+ if (logProgress) {
+ LOGGER.info("Completed spatial index for graph {}", graphNode);
+ }
return Map.entry(graphNode, tree);
}
+ /** Returns a {@link SpatialIndexPerGraph} instance with the configured SRS. */
+ @Override
public SpatialIndexPerGraph call() throws InterruptedException, ExecutionException {
+ try {
+ return callActual();
+ } finally {
+ cleanUp();
+ }
+ }
+
+ protected SpatialIndexPerGraph callActual() throws InterruptedException, ExecutionException {
synchronized (cancelLock) {
if (executorService != null) {
throw new IllegalStateException("Task already running.");
@@ -124,7 +153,9 @@ public class SpatialIndexerComputation {
Entry<Node, STRtree> entry = future.get();
Node graphNode = entry.getKey();
STRtree tree = entry.getValue();
- treeMap.put(graphNode, tree);
+ if (tree != null) {
+ treeMap.put(graphNode, tree);
+ }
}
STRtreePerGraph trees = new STRtreePerGraph();
@@ -133,6 +164,17 @@ public class SpatialIndexerComputation {
return result;
}
+ protected void cleanUp() {
+ if (executorService != null) {
+ executorService.shutdownNow();
+ try {
+ executorService.awaitTermination(5, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ LOGGER.warn("Abandoning an executor service that failed to stop.", e);
+ }
+ }
+ }
+
protected void checkRequestingCancel() {
boolean isCancelled = requestingCancel();
if (isCancelled) {
@@ -140,6 +182,7 @@ public class SpatialIndexerComputation {
}
}
+ @Override
public void abort() {
synchronized (cancelLock) {
requestingCancel.set(true);
@@ -150,17 +193,6 @@ public class SpatialIndexerComputation {
}
}
- public void close() {
- if (executorService != null) {
- executorService.shutdownNow();
- try {
- executorService.awaitTermination(5, TimeUnit.SECONDS);
- } catch (InterruptedException e) {
- throw new RuntimeException("Abandoning an executor serivce that failed to stop.", e);
- }
- }
- }
-
protected void requestCancel() { }
private boolean requestingCancel() {
diff --git a/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/property_functions/GenericSpatialPropertyFunction.java b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/property_functions/GenericSpatialPropertyFunction.java
index 4e9b63245a..12625b59ac 100644
--- a/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/property_functions/GenericSpatialPropertyFunction.java
+++ b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/property_functions/GenericSpatialPropertyFunction.java
@@ -33,7 +33,7 @@ import org.apache.jena.geosparql.spatial.ConvertLatLon;
import org.apache.jena.geosparql.spatial.SearchEnvelope;
import org.apache.jena.geosparql.spatial.SpatialIndex;
import org.apache.jena.geosparql.spatial.SpatialIndexException;
-import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexUtils;
+import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexLib;
import org.apache.jena.graph.Graph;
import org.apache.jena.graph.Node;
import org.apache.jena.graph.Triple;
@@ -66,7 +66,7 @@ public abstract class GenericSpatialPropertyFunction extends PFuncSimpleAndList
@Override
public final QueryIterator execEvaluated(Binding binding, Node subject, Node predicate, PropFuncArg object, ExecutionContext execCxt) {
try {
- spatialIndex = SpatialIndexUtils.retrieve(execCxt);
+ spatialIndex = SpatialIndexLib.retrieve(execCxt);
spatialArguments = extractObjectArguments(predicate, object, spatialIndex.getSrsInfo());
return search(binding, execCxt, subject, spatialArguments.limit);
} catch (SpatialIndexException ex) {
@@ -186,7 +186,7 @@ public abstract class GenericSpatialPropertyFunction extends PFuncSimpleAndList
SearchEnvelope searchEnvelope = spatialArguments.searchEnvelope;
Graph activeGraph = execCxt.getActiveGraph();
- Node graphName = SpatialIndexUtils.unwrapGraphName(activeGraph);
+ Node graphName = SpatialIndexLib.unwrapGraphName(activeGraph);
Collection<Node> features = searchEnvelope.check(spatialIndex, graphName);
Var subjectVar = Var.alloc(subject.getName());
diff --git a/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/task/AbortableThread.java b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/task/AbortableThread.java
deleted file mode 100644
index a32c1d0baa..0000000000
--- a/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/task/AbortableThread.java
+++ /dev/null
@@ -1,199 +0,0 @@
-/*
- * 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.jena.geosparql.spatial.task;
-
-import java.time.Instant;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Objects;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.function.BiConsumer;
-
-import org.apache.jena.geosparql.spatial.task.TaskControl.Registration;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Thread base class that provides a {@link #cancel()} method that can run a custom action
- * (in addition to setting the interrupted flag) as well as a doAfterRun() method
- * for cleaning up after execution (only gets called if there is a prior call to run).
- */
-public abstract class AbortableThread<T>
- extends Thread
-{
- private static final Logger logger = LoggerFactory.getLogger(AbortableThread.class);
-
- private final AtomicBoolean requestingCancel;
- private volatile boolean cancelOnce = false;
- private Object cancelLock = new Object();
-
- protected List<BiConsumer<? super T, Throwable>> completionHandlers = new ArrayList<>();
-
- private boolean isComplete = false;
- private T value = null;
- private Throwable throwable;
-
- private Instant startTime;
- private String statusMessage;
-
- /**
- * Time of the first abort request; null if there is none.
- * Cancel time will not be set if cancel is called after a task's completion.
- */
- private Instant cancelTime;
- private Instant endTime;
-
- // XXX Add native support for CompletableFuture?
-
- public AbortableThread() {
- this(new AtomicBoolean());
- }
-
- public AbortableThread(AtomicBoolean requestingCancel) {
- super();
- this.requestingCancel = Objects.requireNonNull(requestingCancel);
- }
-
- public final boolean isComplete() {
- return isComplete;
- }
-
- public T getValue() {
- return value;
- }
-
- public Throwable getThrowable() {
- return throwable;
- }
-
- public Instant getStartTime() {
- return startTime;
- }
-
- public Instant getEndTime() {
- return endTime;
- }
-
- public Instant getCancelTime() {
- return cancelTime;
- }
-
- protected void setStatusMessage(String statusMessage) {
- this.statusMessage = statusMessage;
- }
-
- public String getStatusMessage() {
- return statusMessage;
- }
-
- @Override
- public final void run() {
- try {
- startTime = Instant.now();
- runInternal();
- } finally {
- synchronized (cancelLock) {
- fireEvents();
- // future.complete(null);
- }
- }
- }
-
- public final void runInternal() {
- try {
- runActual();
- } catch (Throwable t) {
- // t.addSuppressed(new RuntimeException("An error occurred."));
- throwable = t;
- throw new RuntimeException(t);
- } finally {
- endTime = Instant.now();
- isComplete = true;
- doAfterRun();
- }
- }
-
- public abstract void runActual() throws Exception;
-
- /** Returns true iff {@link #cancel()} was called. */
- public final boolean isCancelled() {
- return cancelOnce;
- }
-
- public final void cancel() {
- synchronized (cancelLock) {
- if ( ! cancelOnce && !isComplete ) {
- // Need to set the flags before allowing subclasses to handle requestCancel() in order
- // to prevent a race condition. We want to be sure that calls to have hasNext()/nextBinding()
- // will definitely throw a QueryCancelledException in this class and
- // not allow a situation in which a subclass component thinks it is cancelled,
- // while this class does not.
- if ( requestingCancel != null )
- // Signalling from timeouts
- requestingCancel.set(true);
- cancelOnce = true;
- cancelTime = Instant.now();
- this.requestCancel();
- }
- }
- }
-
- protected void requestCancel() {
- this.interrupt();
- }
-
- protected void doAfterRun() {
- }
-
- protected void fireEvents() {
- for (BiConsumer<? super T, Throwable> handler : completionHandlers) {
- try {
- handler.accept(null, throwable);
- } catch (Throwable t) {
- if (logger.isWarnEnabled()) {
- logger.warn("A task completion handler raised an exception.", t);
- }
- }
- }
- }
-
- /**
- * Registered an action to run on completion of the thread. An action is run at most once.
- *
- * @param action The action is invoked with any thrown exception - null if there was none.
- * @return A registration that can be used to unregister the listener early.
- */
- protected Registration whenComplete(BiConsumer<? super T, Throwable> action) {
- Objects.requireNonNull(action);
- synchronized (cancelLock) {
- boolean isAdded = completionHandlers.add(action);
- if (isAdded) {
- // Immediately resolve if already complete.
- if (isComplete()) {
- action.accept(value, throwable);
- }
- }
- return () -> {
- synchronized (cancelLock) {
- completionHandlers.remove(action);
- }
- };
- }
- }
-}
diff --git a/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/task/TaskControl.java b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/task/BasicTask.java
similarity index 59%
rename from jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/task/TaskControl.java
rename to jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/task/BasicTask.java
index f118c19c29..496f9e153c 100644
--- a/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/task/TaskControl.java
+++ b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/task/BasicTask.java
@@ -18,36 +18,40 @@
package org.apache.jena.geosparql.spatial.task;
-import java.time.Instant;
-import java.util.function.Consumer;
+import org.apache.jena.sparql.engine.iterator.Abortable;
/** An outside view of a running task */
-public interface TaskControl<S> {
- @FunctionalInterface
- interface Registration {
- void dispose();
+public interface BasicTask extends Abortable {
+
+ public interface TaskListener<T extends BasicTask> {
+ void onStateChange(T task);
}
+ TaskState getTaskState();
+
+ /** A label for the task. */
String getLabel();
- S getSource();
+
+ @Override
void abort();
- boolean isAborting();
- boolean isComplete();
- Throwable getThrowable(); // Only meaningful if isComplete is true.
+ long getCreationTime();
+ long getStartTime();
+ long getEndTime();
+ long getAbortTime();
- Instant getStartTime();
- Instant getEndTime();
- Instant getCancelTime();
+ /** If non null, the throwable that is the cause for an exceptional termination of the task. */
+ Throwable getThrowable();
/** Get the last status message of the task. May be null. */
String getStatusMessage();
- /**
- * Registered actions are run only once, then the registration is removed automatically.
- *
- * @param action The action is invoked with any thrown exception - null if there was none.
- * @return A registration that can be used to unregister the listener early.
- */
- Registration whenComplete(Consumer<Throwable> action);
+ /** Whether abort has been called. */
+ // XXX this might be different from whether the task actually transitioned into aborting state.
+ boolean isAborting();
+
+ default boolean isTerminated() {
+ TaskState state = getTaskState();
+ return TaskState.TERMINATED.equals(state);
+ }
}
diff --git a/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/task/TaskControlOverAbortableThread.java b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/task/TaskControlOverAbortableThread.java
deleted file mode 100644
index 986d6d8dbd..0000000000
--- a/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/task/TaskControlOverAbortableThread.java
+++ /dev/null
@@ -1,105 +0,0 @@
-/*
- * 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.jena.geosparql.spatial.task;
-
-import java.time.Instant;
-import java.util.Objects;
-import java.util.function.Consumer;
-
-public class TaskControlOverAbortableThread<S>
- implements TaskControl<S>
-{
- protected String label;
- protected S source;
- protected AbortableThread<?> thread;
-
- public TaskControlOverAbortableThread(String label) {
- super();
- this.label = label;
- }
-
- public void setThread(AbortableThread<?> thread) {
- this.thread = thread;
- }
-
- public void setSource(S source) {
- this.source = source;
- }
-
- @Override
- public String getLabel() {
- return label;
- }
-
- @Override
- public S getSource() {
- return source;
- }
-
- protected AbortableThread<?> requireThread() {
- Objects.requireNonNull(thread);
- return thread;
- }
-
- @Override
- public Instant getStartTime() {
- return requireThread().getStartTime();
- }
-
- @Override
- public Instant getEndTime() {
- return requireThread().getEndTime();
- }
-
- @Override
- public Instant getCancelTime() {
- return requireThread().getCancelTime();
- }
-
- @Override
- public void abort() {
- requireThread().cancel();
- }
-
- @Override
- public boolean isAborting() {
- return requireThread().isCancelled();
- }
-
- @Override
- public boolean isComplete() {
- return !requireThread().isAlive();
- }
-
- @Override
- public Throwable getThrowable() {
- return requireThread().getThrowable();
- }
-
- @Override
- public Registration whenComplete(Consumer<Throwable> action) {
- // Note: The thread's result value is discarded here.
- return requireThread().whenComplete((v, t) -> action.accept(t));
- }
-
- @Override
- public String getStatusMessage() {
- return requireThread().getStatusMessage();
- }
-}
diff --git a/jena-geosparql/src/test/java/org/apache/jena/geosparql/spatial/TestSpatialIndexGraphLookupV2.java b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/task/TaskState.java
similarity index 62%
copy from jena-geosparql/src/test/java/org/apache/jena/geosparql/spatial/TestSpatialIndexGraphLookupV2.java
copy to jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/task/TaskState.java
index 508da2fc79..add2ab9e08 100644
--- a/jena-geosparql/src/test/java/org/apache/jena/geosparql/spatial/TestSpatialIndexGraphLookupV2.java
+++ b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/task/TaskState.java
@@ -15,16 +15,14 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.jena.geosparql.spatial;
-import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexUtils;
-import org.apache.jena.sparql.core.DatasetGraph;
+package org.apache.jena.geosparql.spatial.task;
-public class TestSpatialIndexGraphLookupV2
- extends AbstractSpatialIndexGraphLookpTest
-{
- @Override
- protected SpatialIndex buildSpatialIndex(DatasetGraph dsg, String srsUri) throws SpatialIndexException {
- return SpatialIndexUtils.buildSpatialIndex(dsg, srsUri);
- }
+public enum TaskState {
+ CREATED, // Task object created.
+ STARTING, // Task execution requested, but core task.run() method not yet invoked.
+ RUNNING, // Core task.run() method invoked.
+ ABORTING, // Abort called while not in TERMINATING / TERMINATED state.
+ TERMINATING, // Core task.run() method exited.
+ TERMINATED, // Task cleanup complete. Triggering this event may require a call to close().
}
diff --git a/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/task/TaskThread.java b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/task/TaskThread.java
new file mode 100644
index 0000000000..519778ea58
--- /dev/null
+++ b/jena-geosparql/src/main/java/org/apache/jena/geosparql/spatial/task/TaskThread.java
@@ -0,0 +1,222 @@
+/*
+ * 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.jena.geosparql.spatial.task;
+
+import java.util.Objects;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Thread base class that provides a {@link #abort()} method that can run a custom action
+ * (in addition to setting the interrupted flag) as well as a doAfterRun() method
+ * for cleaning up after execution (only gets called if there is a prior call to run).
+ */
+public abstract class TaskThread
+ extends Thread
+ implements BasicTask
+{
+ private static final Logger logger = LoggerFactory.getLogger(TaskThread.class);
+
+ private final AtomicBoolean requestingCancel;
+ private Object cancelLock = new Object();
+
+ private TaskState state = TaskState.CREATED;
+ private TaskListener<BasicTask> taskListener;
+
+ // protected List<BiConsumer<? super T, Throwable>> completionHandlers = new ArrayList<>();
+
+ private boolean isComplete = false;
+ private Throwable throwable;
+
+ private String label;
+ private String statusMessage;
+
+ /**
+ * Time of the first abort request; null if there is none.
+ * Cancel time will not be set if cancel is called after a task's completion.
+ */
+ private long creationTime = -1;
+ private long startTime = -1; // When when run() is called.
+ private long cancelTime = -1;
+ private long endTime = -1; // Time when runActual returns - but before afterRun() is called.
+
+ // XXX Add native support for CompletableFuture?
+
+ private void updateState(TaskState state) {
+ synchronized (cancelLock) {
+ this.state = state;
+ if (taskListener != null) {
+ try {
+ taskListener.onStateChange(this);
+ } catch (Throwable t) {
+ logger.warn("Listener raised an exception.", t);
+ }
+ }
+ }
+ }
+
+ public TaskThread(String label, TaskListener<BasicTask> taskListener) {
+ this(label, taskListener, new AtomicBoolean());
+ }
+
+ public TaskThread(String label, TaskListener<BasicTask> taskListener, AtomicBoolean requestingCancel) {
+ super();
+ this.label = label;
+ this.taskListener = taskListener;
+ this.requestingCancel = Objects.requireNonNull(requestingCancel);
+
+ this.creationTime = System.currentTimeMillis();
+ updateState(TaskState.CREATED);
+ }
+
+ @Override
+ public String getLabel() {
+ return label;
+ }
+
+ @Override
+ public TaskState getTaskState() {
+ return state;
+ }
+
+ @Override
+ public Throwable getThrowable() {
+ return throwable;
+ }
+
+ @Override
+ public long getCreationTime() {
+ return creationTime;
+ }
+
+ @Override
+ public long getStartTime() {
+ return startTime;
+ }
+
+ @Override
+ public long getEndTime() {
+ return endTime;
+ }
+
+ @Override
+ public long getAbortTime() {
+ return cancelTime;
+ }
+
+ protected void setStatusMessage(String statusMessage) {
+ this.statusMessage = statusMessage;
+ }
+
+ @Override
+ public String getStatusMessage() {
+ return statusMessage;
+ }
+
+ protected void beforeRun() {
+ }
+
+ @Override
+ public final void run() {
+ synchronized (cancelLock) {
+ if (!TaskState.CREATED.equals(state)) {
+ // Task may have already been started or aborted.
+ throw new IllegalStateException("Can only start tasks in CREATED state, but this task is in state: " + state);
+ }
+
+ startTime = System.currentTimeMillis();
+ updateState(TaskState.STARTING);
+ }
+
+ runInternal();
+ }
+
+ public final void runInternal() {
+ try {
+ checkCancelled();
+ beforeRun();
+
+ checkCancelled();
+ runActual();
+ } catch (Throwable t) {
+ throwable = t;
+ updateState(TaskState.TERMINATING);
+ throw new RuntimeException(t);
+ } finally {
+ endTime = System.currentTimeMillis();
+ isComplete = true;
+ try {
+ doAfterRun();
+ } finally {
+ updateState(TaskState.TERMINATED);
+ }
+ }
+ }
+
+ /** The actual "run" method that must be implemented. */
+ public abstract void runActual() throws Exception;
+
+ protected void requestCancel() {
+ // If the task has not been started, then directly transition to terminated with a cancellation exception.
+ // Otherwise transition to aborting state.
+ if (TaskState.CREATED.equals(state)) {
+ this.throwable = new CancellationException();
+ updateState(TaskState.TERMINATED);
+ } else {
+ updateState(TaskState.ABORTING);
+ this.interrupt();
+ }
+ }
+
+ protected void doAfterRun() {
+ }
+
+ /** Raises a {@link CancellationException} if abort was called. */
+ protected void checkCancelled() {
+ if (isAborting()) {
+ throw new CancellationException();
+ }
+ }
+
+ @Override
+ public void abort() {
+ synchronized (cancelLock) {
+ if ( ! requestingCancel.get() ) {
+ requestingCancel.set(true);
+ cancelTime = System.currentTimeMillis();
+ if (!isComplete) {
+ this.requestCancel();
+ }
+ }
+ }
+ }
+
+ @Override
+ public boolean isAborting() {
+ return requestingCancel.get();
+ }
+
+ @Override
+ public boolean isTerminated() {
+ return !isAlive();
+ }
+}
diff --git a/jena-geosparql/src/test/java/org/apache/jena/geosparql/geo/topological/CancelQueryTest.java b/jena-geosparql/src/test/java/org/apache/jena/geosparql/geo/topological/CancelQueryTest.java
index 274a1582dd..ccb0d8d419 100644
--- a/jena-geosparql/src/test/java/org/apache/jena/geosparql/geo/topological/CancelQueryTest.java
+++ b/jena-geosparql/src/test/java/org/apache/jena/geosparql/geo/topological/CancelQueryTest.java
@@ -33,7 +33,7 @@ import org.apache.jena.geosparql.implementation.index.IndexConfiguration;
import org.apache.jena.geosparql.implementation.vocabulary.Geo;
import org.apache.jena.geosparql.spatial.SpatialIndex;
import org.apache.jena.geosparql.spatial.SpatialIndexException;
-import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexUtils;
+import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexLib;
import org.apache.jena.graph.Graph;
import org.apache.jena.graph.NodeFactory;
import org.apache.jena.query.Dataset;
@@ -107,8 +107,8 @@ public class CancelQueryTest {
// create spatial index
if (useIndex){
try {
- SpatialIndex index = SpatialIndexUtils.buildSpatialIndex(ds.asDatasetGraph());
- SpatialIndexUtils.setSpatialIndex(ds, index);
+ SpatialIndex index = SpatialIndexLib.buildSpatialIndex(ds.asDatasetGraph());
+ SpatialIndexLib.setSpatialIndex(ds, index);
} catch (SpatialIndexException e) {
throw new RuntimeException(e);
}
diff --git a/jena-geosparql/src/test/java/org/apache/jena/geosparql/geo/topological/GenericPropertyFunctionTest.java b/jena-geosparql/src/test/java/org/apache/jena/geosparql/geo/topological/GenericPropertyFunctionTest.java
index b20388d4fd..fd405c1b77 100644
--- a/jena-geosparql/src/test/java/org/apache/jena/geosparql/geo/topological/GenericPropertyFunctionTest.java
+++ b/jena-geosparql/src/test/java/org/apache/jena/geosparql/geo/topological/GenericPropertyFunctionTest.java
@@ -41,7 +41,7 @@ import org.apache.jena.geosparql.implementation.index.IndexConfiguration.IndexOp
import org.apache.jena.geosparql.implementation.index.QueryRewriteIndex;
import org.apache.jena.geosparql.implementation.vocabulary.Geo;
import org.apache.jena.geosparql.spatial.SpatialIndexException;
-import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexUtils;
+import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexLib;
import org.apache.jena.graph.Graph;
import org.apache.jena.graph.Node;
import org.apache.jena.graph.NodeFactory;
@@ -79,10 +79,10 @@ public class GenericPropertyFunctionTest {
public static void setUpClass() throws SpatialIndexException {
GeoSPARQLConfig.setup(IndexOption.MEMORY, Boolean.TRUE);
modelGeoSparql = QueryRewriteTestData.addTestDataGeoSparql(ModelFactory.createDefaultModel());
- datasetGeoSparql = SpatialIndexUtils.wrapModel(modelGeoSparql, TEST_SRS_URI);
+ datasetGeoSparql = SpatialIndexLib.wrapModel(modelGeoSparql, TEST_SRS_URI);
modelWgs84 = QueryRewriteTestData.addTestDataWgs84(ModelFactory.createDefaultModel());
- datasetWgs84 = SpatialIndexUtils.wrapModel(modelWgs84, TEST_SRS_URI);
+ datasetWgs84 = SpatialIndexLib.wrapModel(modelWgs84, TEST_SRS_URI);
}
@AfterClass
diff --git a/jena-geosparql/src/test/java/org/apache/jena/geosparql/spatial/SpatialIndexTestData.java b/jena-geosparql/src/test/java/org/apache/jena/geosparql/spatial/SpatialIndexTestData.java
index b917114523..060023a548 100644
--- a/jena-geosparql/src/test/java/org/apache/jena/geosparql/spatial/SpatialIndexTestData.java
+++ b/jena-geosparql/src/test/java/org/apache/jena/geosparql/spatial/SpatialIndexTestData.java
@@ -30,7 +30,7 @@ import org.apache.jena.geosparql.implementation.vocabulary.SRS_URI;
import org.apache.jena.geosparql.spatial.index.v2.STRtreePerGraph;
import org.apache.jena.geosparql.spatial.index.v2.STRtreeUtils;
import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexPerGraph;
-import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexUtils;
+import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexLib;
import org.apache.jena.graph.Node;
import org.apache.jena.query.Dataset;
import org.apache.jena.query.DatasetFactory;
@@ -120,7 +120,7 @@ public class SpatialIndexTestData {
dataset.setDefaultModel(model);
SpatialIndex spatialIndex = createTestIndex();
- SpatialIndexUtils.setSpatialIndex(dataset, spatialIndex);
+ SpatialIndexLib.setSpatialIndex(dataset, spatialIndex);
TEST_DATASET = dataset;
}
diff --git a/jena-geosparql/src/test/java/org/apache/jena/geosparql/spatial/TestSpatialIndexGraphLookupV2.java b/jena-geosparql/src/test/java/org/apache/jena/geosparql/spatial/TestSpatialIndexGraphLookupV2.java
index 508da2fc79..c23f6ff389 100644
--- a/jena-geosparql/src/test/java/org/apache/jena/geosparql/spatial/TestSpatialIndexGraphLookupV2.java
+++ b/jena-geosparql/src/test/java/org/apache/jena/geosparql/spatial/TestSpatialIndexGraphLookupV2.java
@@ -17,7 +17,7 @@
*/
package org.apache.jena.geosparql.spatial;
-import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexUtils;
+import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexLib;
import org.apache.jena.sparql.core.DatasetGraph;
public class TestSpatialIndexGraphLookupV2
@@ -25,6 +25,6 @@ public class TestSpatialIndexGraphLookupV2
{
@Override
protected SpatialIndex buildSpatialIndex(DatasetGraph dsg, String srsUri) throws SpatialIndexException {
- return SpatialIndexUtils.buildSpatialIndex(dsg, srsUri);
+ return SpatialIndexLib.buildSpatialIndex(dsg, srsUri);
}
}
diff --git a/jena-geosparql/src/test/java/org/apache/jena/geosparql/spatial/index/v2/SpatialIndexTest.java b/jena-geosparql/src/test/java/org/apache/jena/geosparql/spatial/index/v2/SpatialIndexTest.java
index 5eb321e604..107d89a5ca 100644
--- a/jena-geosparql/src/test/java/org/apache/jena/geosparql/spatial/index/v2/SpatialIndexTest.java
+++ b/jena-geosparql/src/test/java/org/apache/jena/geosparql/spatial/index/v2/SpatialIndexTest.java
@@ -115,7 +115,7 @@ public class SpatialIndexTest {
Path file = Files.createTempFile("jena-", ".spatial-index");
try {
// Query the built index (no IO)
- SpatialIndexPerGraph indexA = (SpatialIndexPerGraph)SpatialIndexUtils.buildSpatialIndex(dsg);
+ SpatialIndexPerGraph indexA = (SpatialIndexPerGraph)SpatialIndexLib.buildSpatialIndex(dsg);
long itemCountA = indexA.query(envelope, null).size();
Assert.assertEquals(expectedItemCount, itemCountA);
diff --git a/jena-geosparql/src/test/java/org/apache/jena/geosparql/spatial/property_functions/GenericSpatialPropertyFunctionTest.java b/jena-geosparql/src/test/java/org/apache/jena/geosparql/spatial/property_functions/GenericSpatialPropertyFunctionTest.java
index 0b2ff3e6e8..ec16c8000e 100644
--- a/jena-geosparql/src/test/java/org/apache/jena/geosparql/spatial/property_functions/GenericSpatialPropertyFunctionTest.java
+++ b/jena-geosparql/src/test/java/org/apache/jena/geosparql/spatial/property_functions/GenericSpatialPropertyFunctionTest.java
@@ -28,7 +28,7 @@ import org.apache.jena.geosparql.configuration.GeoSPARQLConfig;
import org.apache.jena.geosparql.implementation.vocabulary.SpatialExtension;
import org.apache.jena.geosparql.spatial.SpatialIndexException;
import org.apache.jena.geosparql.spatial.SpatialIndexTestData;
-import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexUtils;
+import org.apache.jena.geosparql.spatial.index.v2.SpatialIndexLib;
import org.apache.jena.query.Dataset;
import org.apache.jena.query.QueryExecution;
import org.apache.jena.query.QueryExecutionFactory;
@@ -203,7 +203,7 @@ public class GenericSpatialPropertyFunctionTest {
Resource geoFeature = ResourceFactory.createResource("http://example.org/GeoFeatureX");
model.add(geoFeature, SpatialExtension.GEO_LAT_PROP, ResourceFactory.createTypedLiteral("0.0", XSDDatatype.XSDfloat));
model.add(geoFeature, SpatialExtension.GEO_LON_PROP, ResourceFactory.createTypedLiteral("0.0", XSDDatatype.XSDfloat));
- Dataset dataset = SpatialIndexUtils.wrapModel(model);
+ Dataset dataset = SpatialIndexLib.wrapModel(model);
String query = "PREFIX spatial: <http://jena.apache.org/spatial#>\n"
+ "\n"