You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@baremaps.apache.org by bc...@apache.org on 2023/09/28 19:44:14 UTC
[incubator-baremaps] 13/35: Improve change importer with copy api
This is an automated email from the ASF dual-hosted git repository.
bchapuis pushed a commit to branch 745-daylight
in repository https://gitbox.apache.org/repos/asf/incubator-baremaps.git
commit 6f4bedaf81763ca6c2128446e086bfa3e581cac7
Author: Bertil Chapuis <bc...@gmail.com>
AuthorDate: Tue Aug 29 00:19:36 2023 +0200
Improve change importer with copy api
---
.run/basemap-workflow.run.xml | 5 +
.../openstreetmap/repository/ChangeImporter.java | 79 ++++++++---
.../java/org/apache/baremaps/workflow/Task.java | 1 +
.../apache/baremaps/workflow/tasks/ImportOsc.java | 91 -------------
.../baremaps/workflow/tasks/ImportOsmChange.java | 56 ++++++++
.../workflow/tasks/UpdateOpenStreetMap.java | 34 +----
basemap/daylight/workflow.js | 144 ++++++++++++---------
7 files changed, 208 insertions(+), 202 deletions(-)
diff --git a/.run/basemap-workflow.run.xml b/.run/basemap-workflow.run.xml
index 714ba264..e1303fa5 100644
--- a/.run/basemap-workflow.run.xml
+++ b/.run/basemap-workflow.run.xml
@@ -10,6 +10,11 @@
<option name="ENABLED" value="true" />
</pattern>
</extension>
+ <extension name="software.aws.toolkits.jetbrains.core.execution.JavaAwsConnectionExtension">
+ <option name="credential" />
+ <option name="region" />
+ <option name="useCurrentConnection" value="false" />
+ </extension>
<method v="2">
<option name="Make" enabled="true" />
</method>
diff --git a/baremaps-core/src/main/java/org/apache/baremaps/openstreetmap/repository/ChangeImporter.java b/baremaps-core/src/main/java/org/apache/baremaps/openstreetmap/repository/ChangeImporter.java
index 979f16b8..6c6a77f9 100644
--- a/baremaps-core/src/main/java/org/apache/baremaps/openstreetmap/repository/ChangeImporter.java
+++ b/baremaps-core/src/main/java/org/apache/baremaps/openstreetmap/repository/ChangeImporter.java
@@ -20,10 +20,14 @@ import org.apache.baremaps.openstreetmap.model.Node;
import org.apache.baremaps.openstreetmap.model.Relation;
import org.apache.baremaps.openstreetmap.model.Way;
import org.apache.baremaps.stream.StreamException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/** A consumer for importing OpenStreetMap changes in a database. */
public class ChangeImporter implements Consumer<Change> {
+ private static final Logger logger = LoggerFactory.getLogger(ChangeImporter.class);
+
private final Repository<Long, Node> nodeRepository;
private final Repository<Long, Way> wayRepository;
private final Repository<Long, Relation> relationRepository;
@@ -47,28 +51,61 @@ public class ChangeImporter implements Consumer<Change> {
/** {@inheritDoc} */
@Override
public void accept(Change change) {
+ var nodes = change.getEntities().stream()
+ .filter(entity -> entity instanceof Node)
+ .map(entity -> (Node) entity)
+ .toList();
+ var ways = change.getEntities().stream()
+ .filter(entity -> entity instanceof Way)
+ .map(entity -> (Way) entity)
+ .toList();
+ var relations = change.getEntities().stream()
+ .filter(entity -> entity instanceof Relation)
+ .map(entity -> (Relation) entity)
+ .toList();
try {
- for (var entity : change.getEntities()) {
- switch (change.getType()) {
- case CREATE:
- case MODIFY:
- if (entity instanceof Node node) {
- nodeRepository.put(node);
- } else if (entity instanceof Way way) {
- wayRepository.put(way);
- } else if (entity instanceof Relation relation) {
- relationRepository.put(relation);
- }
- break;
- case DELETE:
- if (entity instanceof Node node) {
- nodeRepository.delete(node.id());
- } else if (entity instanceof Way way) {
- wayRepository.delete(way.id());
- } else if (entity instanceof Relation relation) {
- relationRepository.delete(relation.id());
- }
- break;
+ switch (change.getType()) {
+ case CREATE -> {
+ if (!nodes.isEmpty()) {
+ logger.info("Creating {} nodes", nodes.size());
+ nodeRepository.copy(nodes);
+ }
+ if (!ways.isEmpty()) {
+ logger.info("Creating {} ways", ways.size());
+ wayRepository.copy(ways);
+ }
+ if (!relations.isEmpty()) {
+ logger.info("Creating {} relations", relations.size());
+ relationRepository.copy(relations);
+ }
+ }
+ case MODIFY -> {
+ if (!nodes.isEmpty()) {
+ logger.info("Modifying {} nodes", nodes.size());
+ nodeRepository.put(nodes);
+ }
+ if (!ways.isEmpty()) {
+ logger.info("Modifying {} ways", ways.size());
+ wayRepository.put(ways);
+ }
+ if (!relations.isEmpty()) {
+ logger.info("Modifying {} relations", relations.size());
+ relationRepository.put(relations);
+ }
+ }
+ case DELETE -> {
+ if (!nodes.isEmpty()) {
+ logger.info("Deleting {} nodes", nodes.size());
+ nodeRepository.delete(nodes.stream().map(Node::id).toList());
+ }
+ if (!ways.isEmpty()) {
+ logger.info("Deleting {} ways", ways.size());
+ wayRepository.delete(ways.stream().map(Way::id).toList());
+ }
+ if (!relations.isEmpty()) {
+ logger.info("Deleting {} relations", relations.size());
+ relationRepository.delete(relations.stream().map(Relation::id).toList());
+ }
}
}
} catch (RepositoryException e) {
diff --git a/baremaps-core/src/main/java/org/apache/baremaps/workflow/Task.java b/baremaps-core/src/main/java/org/apache/baremaps/workflow/Task.java
index 58a0ca71..8a276e93 100644
--- a/baremaps-core/src/main/java/org/apache/baremaps/workflow/Task.java
+++ b/baremaps-core/src/main/java/org/apache/baremaps/workflow/Task.java
@@ -34,6 +34,7 @@ import org.apache.baremaps.workflow.tasks.*;
@JsonSubTypes.Type(value = ExportVectorTiles.class, name = "ExportVectorTiles"),
@JsonSubTypes.Type(value = ImportGeoPackage.class, name = "ImportGeoPackage"),
@JsonSubTypes.Type(value = ImportOsmPbf.class, name = "ImportOsmPbf"),
+ @JsonSubTypes.Type(value = ImportOsmChange.class, name = "ImportOsmChange"),
@JsonSubTypes.Type(value = ImportShapefile.class, name = "ImportShapefile"),
@JsonSubTypes.Type(value = LogMessage.class, name = "LogMessage"),
@JsonSubTypes.Type(value = UnzipFile.class, name = "UnzipFile"),
diff --git a/baremaps-core/src/main/java/org/apache/baremaps/workflow/tasks/ImportOsc.java b/baremaps-core/src/main/java/org/apache/baremaps/workflow/tasks/ImportOsc.java
deleted file mode 100644
index f0aa8285..00000000
--- a/baremaps-core/src/main/java/org/apache/baremaps/workflow/tasks/ImportOsc.java
+++ /dev/null
@@ -1,91 +0,0 @@
-/*
- * Licensed 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.baremaps.workflow.tasks;
-
-import static org.apache.baremaps.stream.ConsumerUtils.consumeThenReturn;
-
-import java.io.BufferedInputStream;
-import java.net.MalformedURLException;
-import java.net.URI;
-import java.net.URL;
-import java.nio.file.Path;
-import java.util.zip.GZIPInputStream;
-import org.apache.baremaps.openstreetmap.function.ChangeEntitiesHandler;
-import org.apache.baremaps.openstreetmap.function.EntityGeometryBuilder;
-import org.apache.baremaps.openstreetmap.function.EntityProjectionTransformer;
-import org.apache.baremaps.openstreetmap.model.Header;
-import org.apache.baremaps.openstreetmap.model.Node;
-import org.apache.baremaps.openstreetmap.model.Relation;
-import org.apache.baremaps.openstreetmap.model.Way;
-import org.apache.baremaps.openstreetmap.postgres.*;
-import org.apache.baremaps.openstreetmap.repository.ChangeImporter;
-import org.apache.baremaps.openstreetmap.repository.Repository;
-import org.apache.baremaps.openstreetmap.state.StateReader;
-import org.apache.baremaps.openstreetmap.xml.XmlChangeReader;
-import org.apache.baremaps.workflow.Task;
-import org.apache.baremaps.workflow.WorkflowContext;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public record ImportOsc(Path file, String database, Integer srid) implements Task {
-
- private static final Logger logger = LoggerFactory.getLogger(ImportOsc.class);
-
- @Override
- public void execute(WorkflowContext context) throws Exception {
- var datasource = context.getDataSource(database);
-
- Repository<Long, Node> nodeRepository = new PostgresNodeRepository(datasource);
- Repository<Long, Way> wayRepository = new PostgresWayRepository(datasource);
- Repository<Long, Relation> relationRepository = new PostgresRelationRepository(datasource);
- execute(
- nodeRepository,
- wayRepository,
- relationRepository,
- srid);
- }
-
- public static void execute(
- Repository<Long, Node> nodeRepository,
- Repository<Long, Way> wayRepository,
- Repository<Long, Relation> relationRepository,
- int srid) throws Exception {
-
- var createGeometry = new EntityGeometryBuilder(coordinateMap, referenceMap);
- var reprojectGeometry = new EntityProjectionTransformer(4326, srid);
- var prepareGeometries = new ChangeEntitiesHandler(createGeometry.andThen(reprojectGeometry));
- var prepareChange = consumeThenReturn(prepareGeometries);
- var saveChange = new ChangeImporter(nodeRepository, wayRepository, relationRepository);
-
- var changeUrl = resolve(replicationUrl, sequenceNumber, "osc.gz");
- try (var changeInputStream =
- new GZIPInputStream(new BufferedInputStream(changeUrl.openStream()))) {
- new XmlChangeReader().stream(changeInputStream).map(prepareChange).forEach(saveChange);
- }
-
- var stateUrl = resolve(replicationUrl, sequenceNumber, "state.txt");
- try (var stateInputStream = new BufferedInputStream(stateUrl.openStream())) {
- var state = new StateReader().state(stateInputStream);
- headerRepository.put(new Header(state.getSequenceNumber(), state.getTimestamp(),
- header.getReplicationUrl(), header.getSource(), header.getWritingProgram()));
- }
- }
-
- public static URL resolve(String replicationUrl, Long sequenceNumber, String extension)
- throws MalformedURLException {
- var s = String.format("%09d", sequenceNumber);
- var uri = String.format("%s/%s/%s/%s.%s", replicationUrl, s.substring(0, 3), s.substring(3, 6),
- s.substring(6, 9), extension);
- return URI.create(uri).toURL();
- }
-}
diff --git a/baremaps-core/src/main/java/org/apache/baremaps/workflow/tasks/ImportOsmChange.java b/baremaps-core/src/main/java/org/apache/baremaps/workflow/tasks/ImportOsmChange.java
new file mode 100644
index 00000000..f32f8c6f
--- /dev/null
+++ b/baremaps-core/src/main/java/org/apache/baremaps/workflow/tasks/ImportOsmChange.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed 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.baremaps.workflow.tasks;
+
+import static org.apache.baremaps.stream.ConsumerUtils.consumeThenReturn;
+
+import java.io.BufferedInputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import org.apache.baremaps.openstreetmap.function.ChangeEntitiesHandler;
+import org.apache.baremaps.openstreetmap.function.EntityGeometryBuilder;
+import org.apache.baremaps.openstreetmap.function.EntityProjectionTransformer;
+import org.apache.baremaps.openstreetmap.postgres.*;
+import org.apache.baremaps.openstreetmap.repository.ChangeImporter;
+import org.apache.baremaps.openstreetmap.xml.XmlChangeReader;
+import org.apache.baremaps.workflow.Task;
+import org.apache.baremaps.workflow.WorkflowContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public record ImportOsmChange(Path file, String database, Integer srid) implements Task {
+
+ private static final Logger logger = LoggerFactory.getLogger(ImportOsmChange.class);
+
+ @Override
+ public void execute(WorkflowContext context) throws Exception {
+ var datasource = context.getDataSource(database);
+
+ var coordinateMap = new PostgresCoordinateMap(datasource);
+ var referenceMap = new PostgresReferenceMap(datasource);
+
+ var nodeRepository = new PostgresNodeRepository(datasource);
+ var wayRepository = new PostgresWayRepository(datasource);
+ var relationRepository = new PostgresRelationRepository(datasource);
+
+ var createGeometry = new EntityGeometryBuilder(coordinateMap, referenceMap);
+ var reprojectGeometry = new EntityProjectionTransformer(4326, srid);
+ var prepareGeometries = new ChangeEntitiesHandler(createGeometry.andThen(reprojectGeometry));
+ var prepareChange = consumeThenReturn(prepareGeometries);
+ var importChange = new ChangeImporter(nodeRepository, wayRepository, relationRepository);
+
+ try (var changeInputStream = new BufferedInputStream(Files.newInputStream(file))) {
+ new XmlChangeReader().stream(changeInputStream).map(prepareChange).forEach(importChange);
+ }
+ }
+}
diff --git a/baremaps-core/src/main/java/org/apache/baremaps/workflow/tasks/UpdateOpenStreetMap.java b/baremaps-core/src/main/java/org/apache/baremaps/workflow/tasks/UpdateOpenStreetMap.java
index 4d0a7092..7cb0f8dd 100644
--- a/baremaps-core/src/main/java/org/apache/baremaps/workflow/tasks/UpdateOpenStreetMap.java
+++ b/baremaps-core/src/main/java/org/apache/baremaps/workflow/tasks/UpdateOpenStreetMap.java
@@ -18,16 +18,11 @@ import java.io.BufferedInputStream;
import java.net.MalformedURLException;
import java.net.URI;
import java.net.URL;
-import java.util.List;
import java.util.zip.GZIPInputStream;
-import org.apache.baremaps.database.collection.DataMap;
import org.apache.baremaps.openstreetmap.function.ChangeEntitiesHandler;
import org.apache.baremaps.openstreetmap.function.EntityGeometryBuilder;
import org.apache.baremaps.openstreetmap.function.EntityProjectionTransformer;
import org.apache.baremaps.openstreetmap.model.Header;
-import org.apache.baremaps.openstreetmap.model.Node;
-import org.apache.baremaps.openstreetmap.model.Relation;
-import org.apache.baremaps.openstreetmap.model.Way;
import org.apache.baremaps.openstreetmap.postgres.PostgresCoordinateMap;
import org.apache.baremaps.openstreetmap.postgres.PostgresHeaderRepository;
import org.apache.baremaps.openstreetmap.postgres.PostgresNodeRepository;
@@ -40,7 +35,6 @@ import org.apache.baremaps.openstreetmap.state.StateReader;
import org.apache.baremaps.openstreetmap.xml.XmlChangeReader;
import org.apache.baremaps.workflow.Task;
import org.apache.baremaps.workflow.WorkflowContext;
-import org.locationtech.jts.geom.Coordinate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -51,33 +45,19 @@ public record UpdateOpenStreetMap(Object database, Integer databaseSrid) impleme
@Override
public void execute(WorkflowContext context) throws Exception {
var datasource = context.getDataSource(database);
- DataMap<Long, Coordinate> coordinateMap = new PostgresCoordinateMap(datasource);
- DataMap<Long, List<Long>> referenceMap = new PostgresReferenceMap(datasource);
- HeaderRepository headerRepository = new PostgresHeaderRepository(datasource);
- Repository<Long, Node> nodeRepository = new PostgresNodeRepository(datasource);
- Repository<Long, Way> wayRepository = new PostgresWayRepository(datasource);
- Repository<Long, Relation> relationRepository = new PostgresRelationRepository(datasource);
- execute(
- coordinateMap,
- referenceMap,
- headerRepository,
- nodeRepository,
- wayRepository,
- relationRepository,
- databaseSrid);
- }
+ var coordinateMap = new PostgresCoordinateMap(datasource);
+ var referenceMap = new PostgresReferenceMap(datasource);
+ var headerRepository = new PostgresHeaderRepository(datasource);
+ var nodeRepository = new PostgresNodeRepository(datasource);
+ var wayRepository = new PostgresWayRepository(datasource);
+ var relationRepository = new PostgresRelationRepository(datasource);
- public static void execute(DataMap<Long, Coordinate> coordinateMap,
- DataMap<Long, List<Long>> referenceMap,
- HeaderRepository headerRepository, Repository<Long, Node> nodeRepository,
- Repository<Long, Way> wayRepository, Repository<Long, Relation> relationRepository,
- int srid) throws Exception {
var header = headerRepository.selectLatest();
var replicationUrl = header.getReplicationUrl();
var sequenceNumber = header.getReplicationSequenceNumber() + 1;
var createGeometry = new EntityGeometryBuilder(coordinateMap, referenceMap);
- var reprojectGeometry = new EntityProjectionTransformer(4326, srid);
+ var reprojectGeometry = new EntityProjectionTransformer(4326, databaseSrid);
var prepareGeometries = new ChangeEntitiesHandler(createGeometry.andThen(reprojectGeometry));
var prepareChange = consumeThenReturn(prepareGeometries);
var saveChange = new ChangeImporter(nodeRepository, wayRepository, relationRepository);
diff --git a/basemap/daylight/workflow.js b/basemap/daylight/workflow.js
index abc12550..51d53e06 100644
--- a/basemap/daylight/workflow.js
+++ b/basemap/daylight/workflow.js
@@ -16,75 +16,93 @@ let config = {
export default {
"steps": [
+ // {
+ // "id": "openstreetmap-data",
+ // "needs": [],
+ // "tasks": [
+ // {
+ // "type": "DownloadUrl",
+ // "url": "https://download.geofabrik.de/europe/liechtenstein-latest.osm.pbf",
+ // "path": "data/data.osm.pbf"
+ // },
+ // {
+ // "type": "ImportOsmPbf",
+ // "file": "data/data.osm.pbf",
+ // "database": config.database,
+ // "databaseSrid": 3857
+ // },
+ // ]
+ // },
+ // {
+ // "id": "openstreetmap-download",
+ // "needs": ["openstreetmap-data"],
+ // "tasks": [
+ // {
+ // "type": "DownloadUrl",
+ // "url": "https://daylight-map-distribution.s3.us-west-1.amazonaws.com/release/v1.29/fb-ml-roads-v1.29.osc.bz2",
+ // "path": "data/roads.osc.bz2"
+ // },
+ // {
+ // "type": "DownloadUrl",
+ // "url": "https://daylight-map-distribution.s3.us-west-1.amazonaws.com/release/v1.29/admin-v1.29.osc.bz2",
+ // "path": "data/admin.osc.bz2"
+ // },
+ // {
+ // "type": "DownloadUrl",
+ // "url": "https://daylight-map-distribution.s3.us-west-1.amazonaws.com/release/v1.29/coastlines-v1.29.tgz",
+ // "path": "data/coastlines.tgz"
+ // },
+ // {
+ // "type": "DownloadUrl",
+ // "url": "https://daylight-map-distribution.s3.us-west-1.amazonaws.com/release/v1.29/preferred-localization-v1.29.tsv",
+ // "path": "data/preferred-localization.tsv"
+ // },
+ // {
+ // "type": "DownloadUrl",
+ // "url": "https://daylight-map-distribution.s3.us-west-1.amazonaws.com/release/v1.29/important-features-v1.29.json",
+ // "path": "data/important-features.json"
+ // },
+ // ]
+ // },
+ // {
+ // "id": "openstreetmap-decompress",
+ // "needs": ["openstreetmap-download"],
+ // "tasks": [
+ // {
+ // "type": "DecompressFile",
+ // "compression": "bzip2",
+ // "source": "data/roads.osc.bz2",
+ // "target": "data/roads.osc"
+ // },
+ // {
+ // "type": "DecompressFile",
+ // "compression": "bzip2",
+ // "source": "data/admin.osc.bz2",
+ // "target": "data/admin.osc"
+ // },
+ // {
+ // "type": "DecompressFile",
+ // "compression": "targz",
+ // "source": "data/coastlines.tgz",
+ // "target": "data/coastlines"
+ // },
+ // ]
+ // },
{
- "id": "openstreetmap-data",
+ "id": "openstreetmap-import",
"needs": [],
"tasks": [
- // {
- // "type": "DownloadUrl",
- // "url": "https://daylight-map-distribution.s3.us-west-1.amazonaws.com/release/v1.29/planet-v1.29.osm.pbf",
- // "path": "data/data.osm.pbf"
- // },
- // {
- // "type": "ImportOsmPbf",
- // "file": "data/data.osm.pbf",
- // "database": config.database,
- // "databaseSrid": 3857
- // },
- ]
- },
- {
- "id": "openstreetmap-download",
- "needs": ["openstreetmap-data"],
- "tasks": [
- {
- "type": "DownloadUrl",
- "url": "https://daylight-map-distribution.s3.us-west-1.amazonaws.com/release/v1.29/fb-ml-roads-v1.29.osc.bz2",
- "path": "data/roads.osc.bz2"
- },
- {
- "type": "DownloadUrl",
- "url": "https://daylight-map-distribution.s3.us-west-1.amazonaws.com/release/v1.29/admin-v1.29.osc.bz2",
- "path": "data/admin.osc.bz2"
- },
- {
- "type": "DownloadUrl",
- "url": "https://daylight-map-distribution.s3.us-west-1.amazonaws.com/release/v1.29/coastlines-v1.29.tgz",
- "path": "data/coastlines.tgz"
- },
- {
- "type": "DownloadUrl",
- "url": "https://daylight-map-distribution.s3.us-west-1.amazonaws.com/release/v1.29/preferred-localization-v1.29.tsv",
- "path": "data/preferred-localization.tsv"
- },
- {
- "type": "DownloadUrl",
- "url": "https://daylight-map-distribution.s3.us-west-1.amazonaws.com/release/v1.29/important-features-v1.29.json",
- "path": "data/important-features.json"
- },
- ]
- },
- {
- "id": "openstreetmap-decompress",
- "needs": ["openstreetmap-download"],
- "tasks": [
- {
- "type": "DecompressFile",
- "compression": "bzip2",
- "source": "data/roads.osc.bz2",
- "target": "data/roads.osc"
- },
{
- "type": "DecompressFile",
- "compression": "bzip2",
- "source": "data/admin.osc.bz2",
- "target": "data/admin.osc"
+ "type": "ImportOsmChange",
+ "file": "data/roads.osc",
+ "database": config.database,
+ "srid": 3857
},
{
- "type": "DecompressFile",
- "compression": "targz",
- "source": "data/coastlines.tgz",
- "target": "data/coastlines"
+ "type": "ImportOsmChange",
+ "file": "data/admin.osc",
+ "database": config.database,
+ "srid": 3857
},
]
},