[BRMO-113] Na verwerken BAG 2 mutaties geometrisch filter toepassen (#…

…2234)

* Remove BAG2 records outside geo filter using sql

* Add CLI option, use transaction, remove only after all mutaties files have been loaded, Dutch messages

* Works for Oracle too

* Configure geo filter in automatisch proces

* Add integration tests using row counts

* Add docs

* Fix ORA-00920 for <23c

* Oracle fix <23c

* Formatting

bag2-loader/README.md

@@ -182,6 +182,27 @@ Let op! Bij het inladen van meerdere gemeentes moeten deze _tegelijkertijd_ opge
 objecten die op de gemeentegrens liggen en dus in meerdere extracten voorkomen slechts één keer in te 
 laden.
 
+#### Geografisch filter toepassen
+
+Met de `--geo-filter` optie kan een geometrie worden opgegeven in WKT formaat. Na het inladen van een stand of het 
+verwerken van mutaties worden dan records die niet snijden met deze geometrie verwijderd en worden records van 
+niet-geografische objecttypen (nummeraanduiding en openbareruimte) waarnaar niet meer gerefereerd wordt verwijderd.
+
+Het geografisch filter wordt opgeslagen in de `brmo_metadata` tabel. Zo kan bij het verwerken van mutaties het 
+geografisch filter ook worden toegepast als de `--geo-filter` optie niet is opgegeven. 
+
+Deze functie is nuttig bij het toepassen van de dagelijkse mutaties die alleen voor heel Nederland beschikbaar zijn, 
+terwijl je alleen geïnteresseerd bent in de BAG van een bepaald gebied.
+
+Het is mogelijk dit filter te gebruiken na het inladen van de stand van heel Nederland. Voor PostGIS zal op een redelijk
+snelle server het verwijderen van records enkele uren kunnen duren, dus hou hier rekening mee. Mogelijk dat het met 
+Oracle te lang duurt. Het is in dat geval verstandig om bij het Kadaster een stand van de gemeentes waarvan de 
+gemeentegrens snijdt met je gewenste geografisch filter te bestellen en in te laden.
+
+Als na het inladen van een stand veel records zijn verwijderd door het geografisch filter, is het verstandig om 
+onderhoud uit te voeren op de BAG tabellen zodat deze geoptimaliseerd zijn door verwijderde records ook fysiek te 
+verwijderen. Met PostGIS kan dit met het `VACUUM` statement, alhoewel dit ook automatisch op de achtergrond plaatsvindt.
+
 ## Database schema
 
 De tabellen die worden aangemaakt bevat historische en toekomstig geldige voorkomens. Gebruik de views zoals 

bag2-loader/src/main/java/nl/b3p/brmo/bag2/loader/cli/BAG2LoadOptions.java

@@ -18,6 +18,9 @@ public class BAG2LoadOptions {
       negatable = true)
   boolean dropIfExists;
 
+  @CommandLine.Option(names = "--geo-filter", paramLabel = "<wkt>")
+  String geoFilter;
+
   @CommandLine.Option(names = "--max-objects", paramLabel = "<number>", hidden = true)
   Integer maxObjects;
 
@@ -45,6 +48,15 @@ public void setDropIfExists(boolean dropIfExists) {
     this.dropIfExists = dropIfExists;
   }
 
+  public String getGeoFilter() {
+    return geoFilter;
+  }
+
+  public BAG2LoadOptions setGeoFilter(String geoFilter) {
+    this.geoFilter = geoFilter;
+    return this;
+  }
+
   public Integer getMaxObjects() {
     return maxObjects;
   }

bag2-loader/src/main/java/nl/b3p/brmo/bag2/loader/cli/BAG2LoaderMain.java

@@ -9,7 +9,7 @@
 
 import static nl.b3p.brmo.bag2.schema.BAG2SchemaMapper.METADATA_TABLE_NAME;
 import static nl.b3p.brmo.bag2.schema.BAG2SchemaMapper.Metadata.CURRENT_TECHNISCHE_DATUM;
-import static nl.b3p.brmo.bag2.schema.BAG2SchemaMapper.Metadata.FILTER_MUTATIES_WOONPLAATS;
+import static nl.b3p.brmo.bag2.schema.BAG2SchemaMapper.Metadata.FILTER_GEOMETRIE;
 import static nl.b3p.brmo.bag2.schema.BAG2SchemaMapper.Metadata.GEMEENTE_CODES;
 import static nl.b3p.brmo.bag2.schema.BAG2SchemaMapper.Metadata.STAND_LOAD_TECHNISCHE_DATUM;
 import static nl.b3p.brmo.bag2.schema.BAG2SchemaMapper.Metadata.STAND_LOAD_TIME;
@@ -25,6 +25,7 @@
 import java.net.http.HttpResponse;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.sql.PreparedStatement;
 import java.text.SimpleDateFormat;
 import java.time.LocalDate;
 import java.util.Date;
@@ -42,7 +43,11 @@
 import nl.b3p.brmo.bag2.schema.BAG2ObjectType;
 import nl.b3p.brmo.bag2.schema.BAG2Schema;
 import nl.b3p.brmo.bag2.schema.BAG2SchemaMapper;
+import nl.b3p.brmo.schema.ObjectType;
+import nl.b3p.brmo.sql.GeometryHandlingPreparedStatementBatch;
 import nl.b3p.brmo.sql.LoggingQueryRunner;
+import nl.b3p.brmo.sql.QueryBatch;
+import nl.b3p.brmo.sql.dialect.OracleDialect;
 import nl.b3p.brmo.util.ResumingInputStream;
 import nl.b3p.brmo.util.http.HttpClientWrapper;
 import nl.b3p.brmo.util.http.HttpStartRangeInputStreamProvider;
@@ -55,6 +60,8 @@
 import org.apache.commons.logging.LogFactory;
 import org.apache.log4j.PropertyConfigurator;
 import org.geotools.util.logging.Logging;
+import org.locationtech.jts.geom.Geometry;
+import org.locationtech.jts.io.WKTReader;
 import picocli.CommandLine;
 import picocli.CommandLine.Command;
 import picocli.CommandLine.ExitCode;
@@ -136,6 +143,19 @@ public int load(
     }
   }
 
+  @Command(name = "apply-geo-filter", sortOptions = false)
+  public int applyGeoFilterCommand(
+      @Mixin BAG2DatabaseOptions dbOptions,
+      @CommandLine.Option(names = "--geo-filter", paramLabel = "<wkt>") String geoFilter)
+      throws Exception {
+    log.info(BAG2LoaderUtils.getUserAgent());
+
+    try (BAG2Database db = getBAG2Database(dbOptions)) {
+      applyGeoFilter(db, new BAG2LoadOptions().setGeoFilter(geoFilter));
+      return ExitCode.OK;
+    }
+  }
+
   public BAG2Database getBAG2Database(BAG2DatabaseOptions dbOptions) throws ClassNotFoundException {
     if (bag2Database == null) {
       bag2Database = new BAG2Database(dbOptions);
@@ -277,8 +297,9 @@ private void loadStandFiles(
         updateMetadata(
             db, loadOptions, true, gemeenteIdentificaties, bagInfo.getStandTechnischeDatum());
       }
-
       db.getConnection().commit();
+
+      applyGeoFilter(db, loadOptions);
     } finally {
       progressReporter.reportTotalSummary();
     }
@@ -378,6 +399,7 @@ private void applyGemeenteMutaties(
       log.info("Mutaties verwerkt, huidige stand technische datum: " + currentTechnischeDatum);
 
     } while (true);
+    applyGeoFilter(db, loadOptions);
   }
 
   private void applyNLMutaties(
@@ -424,6 +446,104 @@ private void applyNLMutaties(
       db.getConnection().commit();
       log.info("Mutaties verwerkt, huidige stand technische datum: " + currentTechnischeDatum);
     } while (true);
+    applyGeoFilter(db, loadOptions);
+  }
+
+  private static final int SRID = 28992;
+
+  public void applyGeoFilter(BAG2Database db, BAG2LoadOptions loadOptions) throws Exception {
+
+    String filterMetadata;
+
+    if (loadOptions.getGeoFilter() != null) {
+      filterMetadata = loadOptions.getGeoFilter();
+      try {
+        new WKTReader().read(filterMetadata);
+      } catch (Exception e) {
+        log.error("Ongeldige WKT gespecificeerd voor geometrie-filter", e);
+        return;
+      }
+      db.setMetadataValue(FILTER_GEOMETRIE, filterMetadata);
+    } else {
+      filterMetadata = db.getMetadata(FILTER_GEOMETRIE);
+    }
+
+    if (filterMetadata == null) {
+      return;
+    }
+    Geometry geometry;
+    try {
+      geometry = new WKTReader().read(filterMetadata);
+      geometry.setSRID(SRID);
+    } catch (Exception e) {
+      log.error(
+          String.format(
+              "Geometrie-filter niet toegepast, fout bij parsen %s als WKT, waarde: \"%s\"",
+              FILTER_GEOMETRIE, filterMetadata),
+          e);
+      return;
+    }
+    log.info("Verwijderen records die niet binnen geometrie-filter vallen...");
+
+    BAG2SchemaMapper schemaMapper = BAG2SchemaMapper.getInstance();
+    BAG2Schema bag2Schema = BAG2Schema.getInstance();
+
+    db.getConnection().setAutoCommit(false);
+
+    String[] objectTypes =
+        new String[] {"Ligplaats", "Standplaats", "Pand", "Verblijfsobject", "Woonplaats"};
+
+    for (String objectTypeName : objectTypes) {
+      ObjectType objectType = bag2Schema.getObjectTypeByName(objectTypeName);
+      String tableName = schemaMapper.getTableNameForObjectType(objectType, "");
+      String geoCondition =
+          db.getDialect() instanceof OracleDialect
+              ? "st_intersects(?, geometrie) = 'FALSE'"
+              : "not st_intersects(geometrie, ?)"; // Also works for Oracle 23c, not < 23c
+      String sql =
+          String.format(
+              """
+                delete from %s
+                where identificatie in (
+                  select identificatie from %s
+                  where eindgeldigheid is null and tijdstipinactief is null
+                  and %s
+                )""",
+              tableName, tableName, geoCondition);
+      try (QueryBatch queryBatch =
+          new GeometryHandlingPreparedStatementBatch(
+              db.getConnection(), sql, 1, db.getDialect(), new Boolean[] {true}, false)) {
+        log.info("Verwijderen van records voor " + objectType.getName());
+        queryBatch.addBatch(new Object[] {geometry});
+      }
+    }
+
+    // Don't use getObjectTypeByName(), because heeftAlsNevenadres join tables need to be directly
+    // in sql literal anyway
+    String sql =
+        """
+        delete from nummeraanduiding n
+        where not exists (select 1 from verblijfsobject v where v.heeftalshoofdadres = n.identificatie)
+        and not exists (select 1 from verblijfsobject_nevenadres vn where vn.heeftalsnevenadres = n.identificatie)
+        and not exists (select 1 from ligplaats l where l.heeftalshoofdadres = n.identificatie)
+        and not exists (select 1 from ligplaats_nevenadres ln where ln.heeftalsnevenadres = n.identificatie)
+        and not exists (select 1 from standplaats s where s.heeftalshoofdadres = n.identificatie)
+        and not exists (select 1 from standplaats_nevenadres sn where sn.heeftalsnevenadres = n.identificatie)
+        """;
+    try (PreparedStatement ps = db.getConnection().prepareStatement(sql)) {
+      log.info("Verwijderen niet-gerefereerde nummeraanduiding-records");
+      ps.executeUpdate();
+    }
+    sql =
+        """
+        delete from openbareruimte o where not exists (select 1 from nummeraanduiding where ligtaan = o.identificatie)
+        """;
+    try (PreparedStatement ps = db.getConnection().prepareStatement(sql)) {
+      log.info("Verwijderen niet-gerefereerde openbareruimte-records");
+      ps.executeUpdate();
+    }
+    db.getConnection().commit();
+    log.info("Klaar met verwijderen records buiten geometrie-filter");
   }
 
   private void createKeysAndIndexes(
@@ -633,7 +753,6 @@ private void updateMetadata(
           STAND_LOAD_TIME, new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
       db.setMetadataValue(STAND_LOAD_TECHNISCHE_DATUM, df.format(standTechnischeDatum));
       db.setMetadataValue(GEMEENTE_CODES, String.join(",", gemeenteIdentificaties));
-      db.setMetadataValue(FILTER_MUTATIES_WOONPLAATS, "false");
     }
     db.setMetadataValue(CURRENT_TECHNISCHE_DATUM, df.format(standTechnischeDatum));
   }

bag2-loader/src/main/java/nl/b3p/brmo/bag2/schema/BAG2SchemaMapper.java

@@ -29,7 +29,7 @@ public enum Metadata {
     STAND_LOAD_TECHNISCHE_DATUM,
     CURRENT_TECHNISCHE_DATUM,
     GEMEENTE_CODES,
-    FILTER_MUTATIES_WOONPLAATS;
+    FILTER_GEOMETRIE;
 
     public String getDbKey() {
       return "bag2_" + this.name().toLowerCase();

bag2-loader/src/test/java/nl/b3p/brmo/bag2/loader/BAGLoaderDatabaseIntegrationTest.java

@@ -6,6 +6,7 @@
  */
 package nl.b3p.brmo.bag2.loader;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 import static org.junit.jupiter.api.Assumptions.assumeFalse;
@@ -15,11 +16,14 @@
 import java.sql.DriverManager;
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.util.Map;
 import java.util.Objects;
 import nl.b3p.brmo.bag2.loader.cli.BAG2DatabaseOptions;
 import nl.b3p.brmo.bag2.loader.cli.BAG2LoadOptions;
 import nl.b3p.brmo.bag2.loader.cli.BAG2LoaderMain;
+import nl.b3p.brmo.bag2.schema.BAG2SchemaMapper;
 import nl.b3p.brmo.sql.LoggingQueryRunner;
+import nl.b3p.brmo.sql.dialect.PostGISDialect;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -241,22 +245,86 @@ void testCompareDataSetWoonplaatsMutatie() throws Exception {
     compareDataSet(new String[] {"woonplaats"}, "bag2-woonplaats-gemuteerd");
   }
 
+  private void assertRowCountEquals(Map<String, Integer> expectedRowCountsByTable)
+      throws SQLException {
+    for (Map.Entry<String, Integer> entry : expectedRowCountsByTable.entrySet()) {
+      String table = entry.getKey();
+      Integer count = entry.getValue();
+      assertEquals(
+          count.intValue(), bag.getRowCount(tableQualifierPrefix + table), "Table " + table);
+    }
+  }
+
+  @Test
+  @Order(3)
+  void testLoadGemeenteStand() throws Exception {
+    loadBAGResourceFile(testFileName);
+
+    assertRowCountEquals(
+        Map.ofEntries(
+            Map.entry("ligplaats", 432),
+            Map.entry("ligplaats_nevenadres", 43),
+            Map.entry("standplaats", 54),
+            Map.entry("standplaats_nevenadres", 1),
+            Map.entry("verblijfsobject", 18506),
+            Map.entry("verblijfsobject_nevenadres", 1),
+            Map.entry("verblijfsobject_maaktdeeluitvan", 18562),
+            Map.entry("verblijfsobject_gebruiksdoel", 19221),
+            Map.entry("pand", 10000),
+            Map.entry("nummeraanduiding", 10005),
+            Map.entry("openbareruimte", 761),
+            Map.entry("woonplaats", 14)));
+  }
+
+  @Test
+  @Order(4)
+  @SkipDropTables
+  void testFilterStandGeographicFilter() throws Exception {
+    try {
+      String geoFilter =
+          "Polygon ((128193 464844, 128500 464913, 128515 464915, 128520 464917, 129239 465079, 130329 465177, 130354 465421, 130965 465408, 131033 465015, 129235 464729, 129419 464062, 129103 463870, 128825 463998, 128291 463720, 128193 464844))";
+      new BAG2LoaderMain()
+          .applyGeoFilter(bag2Database, new BAG2LoadOptions().setGeoFilter(geoFilter));
+    } catch (Exception e) {
+      fail("Toepassen geografisch filter op BAG is mislukt", e);
+    }
+
+    assertRowCountEquals(
+        Map.ofEntries(
+            Map.entry("ligplaats", 3),
+            Map.entry("ligplaats_nevenadres", 0),
+            Map.entry("standplaats", 6),
+            Map.entry("standplaats_nevenadres", 0),
+            Map.entry("verblijfsobject", 1900),
+            Map.entry("verblijfsobject_nevenadres", 0),
+            Map.entry("verblijfsobject_maaktdeeluitvan", 1912),
+            Map.entry("verblijfsobject_gebruiksdoel", 1952),
+            Map.entry("pand", 2595),
+            Map.entry("nummeraanduiding", 838),
+            Map.entry("openbareruimte", 32),
+            Map.entry("woonplaats", 1)));
+  }
+
   // Leave this as last integration test case, so the BAG tables loaded by it can be used to test
   // the create view scripts
   @Test
   @Order(Integer.MAX_VALUE)
   void testStandAllTablesAndViewsHaveRows() throws Exception {
+    if (bag2Database.getDialect() instanceof PostGISDialect) {
+      // For Oracle, BRMO_METADATA table will have been dropped, for PostGIS not because it's in
+      // public schema
+      bag2Database.setMetadataValue(BAG2SchemaMapper.Metadata.FILTER_GEOMETRIE, null);
+    }
     loadBAGResourceFile(testFileName);
 
     // check tables
     for (String t : BAGTABLES) {
       // omdat sommige BAG tabellen ook in RSGB schema zitten bag qualifier gebruiken
-      t = tableQualifierPrefix + t;
-      assertTrue(bag.getRowCount(t) > 0, "Onverwacht lege tabel: " + t);
+      assertTrue(bag.getRowCount(tableQualifierPrefix + t) > 0, "Onverwacht lege tabel: " + t);
     }
     // check views
     for (String t : BAGACTUEELVIEWS) {
-      assertTrue(bag.getRowCount(t) > 0, "Onverwacht lege view: " + t);
+      assertTrue(bag.getRowCount(tableQualifierPrefix + t) > 0, "Onverwacht lege view: " + t);
     }
   }
 }

bgt-loader/src/main/java/nl/b3p/brmo/sql/QueryBatch.java

@@ -8,7 +8,7 @@
 
 import java.sql.SQLException;
 
-public interface QueryBatch {
+public interface QueryBatch extends AutoCloseable {
   boolean addBatch(Object[] params) throws Exception;
 
   void executeBatch() throws Exception;

brmo-service/src/main/java/nl/b3p/brmo/service/scanner/BAG2MutatieProcesRunner.java

@@ -121,6 +121,7 @@ public void close() {
 
         // Use defaults
         BAG2LoadOptions loadOptions = new BAG2LoadOptions();
+        loadOptions.setGeoFilter(ClobElement.nullSafeGet(config.getConfig().get("geo-filter")));
         BAG2DatabaseOptions dbOptions = new BAG2DatabaseOptions();
 
         listener.updateStatus(PROCESSING.toString());