View Javadoc
1   package de.dlr.shepard.data.spatialdata.repositories;
2   
3   import de.dlr.shepard.common.util.JsonConverter;
4   import de.dlr.shepard.data.spatialdata.io.FilterCondition;
5   import de.dlr.shepard.data.spatialdata.model.SpatialDataPoint;
6   import io.micrometer.core.annotation.Timed;
7   import io.quarkus.hibernate.orm.PersistenceUnit;
8   import jakarta.enterprise.context.RequestScoped;
9   import jakarta.enterprise.inject.Instance;
10  import jakarta.persistence.EntityManager;
11  import java.util.List;
12  import java.util.Locale;
13  import java.util.Map;
14  import org.eclipse.microprofile.config.inject.ConfigProperty;
15  import org.locationtech.jts.geom.Coordinate;
16  
17  @RequestScoped
18  public class SpatialDataPointRepository {
19  
20    private static final int INSERT_BATCH_SIZE = 20000;
21    public static final String SPATIAL_TABLE_NAME = "spatial_data_points";
22    public static final String SPATIAL_COLUMN_CONTAINER_ID = "container_id";
23    public static final String SPATIAL_COLUMN_TIME = "time";
24    public static final String SPATIAL_COLUMN_POSITION = "position";
25    public static final String SPATIAL_COLUMN_METADATA = "metadata";
26    public static final String SPATIAL_COLUMN_MEASUREMENTS = "measurements";
27  
28    private static final String[] ALL_COLUMNS_STRING = new String[] { "*" };
29  
30    @PersistenceUnit("spatial")
31    Instance<EntityManager> entityManager;
32  
33    @ConfigProperty(name = "shepard.spatial-data.enabled")
34    boolean spatialDataEnabled;
35  
36    EntityManager getEntityManager() {
37      if (spatialDataEnabled) {
38        return entityManager.get();
39      } else {
40        throw new IllegalStateException("Spatial Data is disabled yet SpatialDataPointRepository is used!");
41      }
42    }
43  
44    @Timed(value = "shepard.spatial-data.insert")
45    public int insert(long containerId, SpatialDataPoint data) {
46      var sql = new NativeInsertStatementBuilder()
47        .insert(
48          SPATIAL_TABLE_NAME,
49          new String[] {
50            SPATIAL_COLUMN_CONTAINER_ID,
51            SPATIAL_COLUMN_TIME,
52            SPATIAL_COLUMN_POSITION,
53            SPATIAL_COLUMN_METADATA,
54            SPATIAL_COLUMN_MEASUREMENTS,
55          }
56        )
57        .addValues(
58          String.format(
59            Locale.US,
60            "%d, '%s', ST_MakePoint(%f, %f, %f), CAST('%s' AS JSONB), CAST('%s' AS JSONB)",
61            containerId,
62            data.getTime(),
63            data.getPosition().getCoordinate().x,
64            data.getPosition().getCoordinate().y,
65            data.getPosition().getCoordinate().z,
66            JsonConverter.convertToString(data.getMetadata()),
67            JsonConverter.convertToString(data.getMeasurements())
68          )
69        )
70        .build();
71  
72      var query = getEntityManager().createNativeQuery(sql);
73      var resultCount = query.executeUpdate();
74      if (resultCount <= 0) throw new DbUpdateException("SpatialData was not stored in database.");
75      return resultCount;
76    }
77  
78    @Timed(value = "shepard.spatial-data.insert-many")
79    public int insert(long containerId, SpatialDataPoint[] data) {
80      var allResultCount = 0;
81      var sql = new NativeInsertStatementBuilder()
82        .insert(
83          SPATIAL_TABLE_NAME,
84          new String[] {
85            SPATIAL_COLUMN_CONTAINER_ID,
86            SPATIAL_COLUMN_TIME,
87            SPATIAL_COLUMN_POSITION,
88            SPATIAL_COLUMN_METADATA,
89            SPATIAL_COLUMN_MEASUREMENTS,
90          }
91        );
92  
93      for (int i = 0; i < data.length; i += INSERT_BATCH_SIZE) {
94        int currentLimit = Math.min(i + INSERT_BATCH_SIZE, data.length);
95        for (int j = i; j < currentLimit; j++) {
96          sql.addValues(
97            String.format(
98              Locale.US,
99              "%d, '%s', ST_MakePoint(%f, %f, %f), CAST('%s' AS JSONB), CAST('%s' AS JSONB)",
100             containerId,
101             data[j].getTime(),
102             data[j].getPosition().getCoordinate().x,
103             data[j].getPosition().getCoordinate().y,
104             data[j].getPosition().getCoordinate().z,
105             JsonConverter.convertToString(data[j].getMetadata()),
106             JsonConverter.convertToString(data[j].getMeasurements())
107           )
108         );
109       }
110       var query = getEntityManager().createNativeQuery(sql.build());
111       allResultCount += query.executeUpdate();
112     }
113     return allResultCount;
114   }
115 
116   /**
117    * @param containerId to be used for deletion
118    * @return the number of rows deleted
119    */
120   @Timed(value = "shepard.spatial-data.delete-by-container")
121   public int deleteByContainerId(long containerId) {
122     return getEntityManager()
123       .createNativeQuery(
124         "DELETE FROM %s WHERE %s=:containerId;".formatted(SPATIAL_TABLE_NAME, SPATIAL_COLUMN_CONTAINER_ID)
125       )
126       .setParameter("containerId", containerId)
127       .executeUpdate();
128   }
129 
130   @Timed(value = "shepard.spatial-data.get-by-container")
131   @SuppressWarnings("unchecked")
132   public List<SpatialDataPoint> getByContainerId(long containerId) {
133     var query = new NativeQueryStringBuilder()
134       .select(SPATIAL_TABLE_NAME, ALL_COLUMNS_STRING)
135       .addWhereCondition(SPATIAL_COLUMN_CONTAINER_ID, containerId)
136       .build();
137 
138     return getEntityManager().createNativeQuery(query, SpatialDataPoint.class).getResultList();
139   }
140 
141   @Timed(value = "shepard.spatial-data.query-by-bounding-box")
142   @SuppressWarnings("unchecked")
143   public List<SpatialDataPoint> get(
144     long containerId,
145     Long timestampStart,
146     Long timestampEnd,
147     Map<String, Object> metadataFilter,
148     List<FilterCondition> measurementsFilter,
149     Integer limit,
150     Integer skip
151   ) {
152     var queryBuilder = new NativeQueryStringBuilder()
153       .select(SPATIAL_TABLE_NAME, ALL_COLUMNS_STRING)
154       .addWhereCondition(SPATIAL_COLUMN_CONTAINER_ID, containerId)
155       .addTimeCondition(SPATIAL_COLUMN_TIME, timestampStart, timestampEnd)
156       .addJsonContainsCondition(SPATIAL_COLUMN_METADATA, metadataFilter)
157       .addJsonFilterConditions(SPATIAL_COLUMN_MEASUREMENTS, measurementsFilter)
158       .addSkipClause(skip)
159       .addLimitClause(limit);
160 
161     var query = getEntityManager().createNativeQuery(queryBuilder.build(), SpatialDataPoint.class);
162     queryBuilder.getQueryParameters().forEach(query::setParameter);
163     return query.getResultList();
164   }
165 
166   /**
167    * Create an (axis-aligned) bounding box query request for spatial data.
168    * The request uses the '&&&'' indexed operator, that acts similar to the
169    * ST_Intersects function.
170    * If the point is part of the bounding box (i.e., on a bounding box corner) the
171    * bounding box check returns true
172    */
173   @Timed(value = "shepard.spatial-data.query-by-bounding-box")
174   @SuppressWarnings("unchecked")
175   public List<SpatialDataPoint> getByBoundingBox(
176     long containerId,
177     Coordinate bottomLeft,
178     Coordinate topRight,
179     Long timestampStart,
180     Long timestampEnd,
181     Map<String, Object> metadataFilter,
182     List<FilterCondition> measurementsFilter,
183     Integer limit,
184     Integer skip
185   ) {
186     var queryBuilder = new NativeQueryStringBuilder()
187       .select(SPATIAL_TABLE_NAME, ALL_COLUMNS_STRING)
188       .addWhereCondition(SPATIAL_COLUMN_CONTAINER_ID, containerId)
189       .addTimeCondition(SPATIAL_COLUMN_TIME, timestampStart, timestampEnd)
190       .addJsonContainsCondition(SPATIAL_COLUMN_METADATA, metadataFilter)
191       .addJsonFilterConditions(SPATIAL_COLUMN_MEASUREMENTS, measurementsFilter)
192       .addAABBGeometryCondition(bottomLeft.x, bottomLeft.y, bottomLeft.z, topRight.x, topRight.y, topRight.z)
193       .addSkipClause(skip)
194       .addLimitClause(limit);
195 
196     var query = getEntityManager().createNativeQuery(queryBuilder.build(), SpatialDataPoint.class);
197     queryBuilder.getQueryParameters().forEach(query::setParameter);
198     return query.getResultList();
199   }
200 
201   @SuppressWarnings("unchecked")
202   @Timed(value = "shepard.spatial-data.query-by-bounding-sphere")
203   public List<SpatialDataPoint> getByBoundingSphere(
204     long containerId,
205     Coordinate coordinate,
206     double radius,
207     Long timestampStart,
208     Long timestampEnd,
209     Map<String, Object> metadataFilter,
210     List<FilterCondition> measurementsFilter,
211     Integer limit,
212     Integer skip
213   ) {
214     var queryBuilder = new NativeQueryStringBuilder()
215       .select(SPATIAL_TABLE_NAME, ALL_COLUMNS_STRING)
216       .addWhereCondition(SPATIAL_COLUMN_CONTAINER_ID, containerId)
217       .addTimeCondition(SPATIAL_COLUMN_TIME, timestampStart, timestampEnd)
218       .addJsonContainsCondition(SPATIAL_COLUMN_METADATA, metadataFilter)
219       .addJsonFilterConditions(SPATIAL_COLUMN_MEASUREMENTS, measurementsFilter)
220       .addBSGeometryCondition(coordinate.x, coordinate.y, coordinate.z, radius)
221       .addSkipClause(skip)
222       .addLimitClause(limit);
223 
224     var query = getEntityManager().createNativeQuery(queryBuilder.build(), SpatialDataPoint.class);
225     queryBuilder.getQueryParameters().forEach(query::setParameter);
226     return query.getResultList();
227   }
228 
229   /**
230    * Runs a k-nearest-neighbor search on the spatial data.
231    *
232    * @param coordinate - Starting point for the KNN search
233    * @param k          - number of returned points
234    */
235   @SuppressWarnings("unchecked")
236   @Timed(value = "shepard.spatial-data.query-by-knn")
237   public List<SpatialDataPoint> getByKNN(
238     long containerId,
239     Coordinate coordinate,
240     int k,
241     Long timestampStart,
242     Long timestampEnd,
243     Map<String, Object> metadataFilter,
244     List<FilterCondition> measurementsFilter
245   ) {
246     var queryBuilder = new NativeQueryStringBuilder()
247       .select(SPATIAL_TABLE_NAME, ALL_COLUMNS_STRING)
248       .addWhereCondition(SPATIAL_COLUMN_CONTAINER_ID, containerId)
249       .addTimeCondition(SPATIAL_COLUMN_TIME, timestampStart, timestampEnd)
250       .addJsonContainsCondition(SPATIAL_COLUMN_METADATA, metadataFilter)
251       .addJsonFilterConditions(SPATIAL_COLUMN_MEASUREMENTS, measurementsFilter)
252       .addKNNGeometryCondition(coordinate.x, coordinate.y, coordinate.z, k);
253 
254     var query = getEntityManager().createNativeQuery(queryBuilder.build(), SpatialDataPoint.class);
255     queryBuilder.getQueryParameters().forEach(query::setParameter);
256     return query.getResultList();
257   }
258 }