FFmpeg nullcheck and synchronization to prevent native memory exception

persistence/sensorhub-datastore-postgis/src/main/java/org/sensorhub/impl/datastore/postgis/ConnectionManager.java

@@ -53,7 +53,8 @@ private HikariDataSource createHikariDataSource() {
         config.setJdbcUrl("jdbc:postgresql://" + url + "/" + dbName);
         config.setUsername(login);
         config.setPassword(password);
-        config.setMaximumPoolSize(5);
+        // TODO: Set using postgis config
+        config.setMaximumPoolSize(50);
         config.setConnectionTimeout(1000 * 60 * 5); // 5 minutes
 
 //                        config.setMaximumPoolSize(200_000);

persistence/sensorhub-datastore-postgis/src/main/java/org/sensorhub/impl/datastore/postgis/builder/IteratorResultSet.java

@@ -29,10 +29,14 @@
 public class IteratorResultSet<T> implements Iterator<T> {
     private static final Logger logger = LoggerFactory.getLogger(IteratorResultSet.class);
 
-    private long limit = Long.MAX_VALUE;
+    private long internalLimit;
+
+    private long filterLimit;
 
     private long offset = 0;
 
+    private long deliveredCount = 0;
+
     private String query;
 
     private ConnectionManager connectionManager;
@@ -48,71 +52,136 @@ public class IteratorResultSet<T> implements Iterator<T> {
 
     public IteratorResultSet(String query,
                              ConnectionManager connectionManager,
-                             long limit,
+                             long internalLimit,
+                             long filterLimit,
                              Function<ResultSet, T> parsingFn,
-                             Function<T, Boolean> predicateValidator
+                             Function<T, Boolean> predicateValidator,
+                             boolean hasValuePredicate
     ) {
-        this.query = query;
-        this.limit = limit;
+        this.query = !hasValuePredicate ? query : removeSqlLimit(query);
+        this.internalLimit = internalLimit;
+        this.filterLimit = filterLimit;
         this.parsingFn = parsingFn;
         this.connectionManager = connectionManager;
         this.predicateValidator = predicateValidator;
-        this.useInternalLimit = !query.contains("LIMIT");
+        this.useInternalLimit = !query.contains("LIMIT") || hasValuePredicate;
     }
 
     @Override
     public boolean hasNext() {
+        if (deliveredCount >= filterLimit) {
+            return false;
+        }
         if(!records.isEmpty()) {
             return true;
         }
         if (ended) {
             return false;
         }
-        while(records.isEmpty() && !ended) {
+        while(records.isEmpty() && !ended && deliveredCount < filterLimit   ) {
             this.makeRequest();
         }
         return !records.isEmpty();
     }
 
     private String getQuery() {
         if(useInternalLimit) {
-            return query + " LIMIT " + limit + " OFFSET " + offset;
+            return query + " LIMIT " + internalLimit + " OFFSET " + offset;
         } else {
             // limit set by the filter itself
             return query + " OFFSET " + offset;
         }
+    }
 
+    private String removeSqlLimit(String sql) {
+        return sql.replaceAll("(?i)\\s+LIMIT\\s+\\d+(\\s+OFFSET\\s+\\d+)?", "");
     }
 
+
     @Override
     public T next() {
-        return records.poll();
+        T value = records.poll();
+        if (value != null) {
+            deliveredCount++;
+        }
+        return value;
     }
 
     private void makeRequest() {
-        long countRes = 0;
+        long fetchedFromDb = 0;
+
         try (Connection connection = connectionManager.getConnection()) {
-            try(Statement statement = connection.createStatement()) {
+            try (Statement statement = connection.createStatement()) {
                 String nextQuery = getQuery();
-                if(logger.isDebugEnabled()) {
+                if (logger.isDebugEnabled()) {
                     logger.debug(nextQuery);
                 }
-                try (ResultSet resultSet = statement.executeQuery(nextQuery)){
+                try (ResultSet resultSet = statement.executeQuery(nextQuery)) {
                     while (resultSet.next()) {
-                        countRes++;
-                        T res = this.parsingFn.apply(resultSet);
-                        if(predicateValidator.apply(res)) {
+                        fetchedFromDb++;
+
+                        if (deliveredCount + records.size() >= filterLimit) {
+                            ended = true;
+                            break;
+                        }
+
+                        T res = parsingFn.apply(resultSet);
+
+                        if (predicateValidator.apply(res)) {
                             records.add(res);
+                            if (deliveredCount + records.size() >= filterLimit) {
+                                ended = true;
+                                break;
+                            }
                         }
                     }
-                    offset += limit;
                 }
             }
-            if(countRes == 0 || countRes < limit) {
+
+            // Move offset only when internal limit is active
+            if (useInternalLimit) {
+                offset += internalLimit;
+            } else {
+                // If SQL LIMIT is controlling the batch size, we must rely on SQL side
+                offset += filterLimit > 0 ? filterLimit : internalLimit;
+            }
+
+            // If DB returned fewer rows than internalLimit, this is the final page
+            if (fetchedFromDb == 0 ||
+                    (useInternalLimit && fetchedFromDb < internalLimit)) {
                 ended = true;
             }
+
         } catch (SQLException e) {
             throw new RuntimeException(e);
         }
     }
+
+
+//    private void makeRequest() {
+//        long countRes = 0;
+//        try (Connection connection = connectionManager.getConnection()) {
+//            try(Statement statement = connection.createStatement()) {
+//                String nextQuery = getQuery();
+//                if(logger.isDebugEnabled()) {
+//                    logger.debug(nextQuery);
+//                }
+//                try (ResultSet resultSet = statement.executeQuery(nextQuery)){
+//                    while (resultSet.next()) {
+//                        countRes++;
+//                        T res = this.parsingFn.apply(resultSet);
+//                        if(predicateValidator.apply(res)) {
+//                            records.add(res);
+//                        }
+//                    }
+//                    offset += limit;
+//                }
+//            }
+//            if(countRes == 0 || countRes < limit) {
+//                ended = true;
+//            }
+//        } catch (SQLException e) {
+//            throw new RuntimeException(e);
+//        }
+//    }
 }

persistence/sensorhub-datastore-postgis/src/main/java/org/sensorhub/impl/datastore/postgis/builder/QueryBuilder.java

@@ -26,7 +26,7 @@
 import org.sensorhub.api.datastore.system.ISystemDescStore;
 
 public abstract class QueryBuilder {
-    private final String tableName;
+    protected final String tableName;
 
     protected IDataStreamStore dataStreamStore;
     protected IProcedureStore procedureStore;

persistence/sensorhub-datastore-postgis/src/main/java/org/sensorhub/impl/datastore/postgis/builder/QueryBuilderCommandStatusStore.java

@@ -20,10 +20,15 @@
 
 import java.util.Set;
 
-import static org.sensorhub.api.datastore.command.ICommandStatusStore.CommandStatusField.COMMAND_ID;
+import static org.sensorhub.api.datastore.command.ICommandStatusStore.CommandStatusField.*;
 
 public class QueryBuilderCommandStatusStore extends QueryBuilder {
     public final static String COMMAND_TABLE_NAME = "commandstatus";
+
+    protected static final String PROGRESS = "progress";
+    protected static final String MESSAGE = "message";
+    protected static final String RESULT = "result";
+
     public QueryBuilderCommandStatusStore() {
         super(COMMAND_TABLE_NAME);
     }
@@ -37,17 +42,32 @@ public String createTableQuery() {
         return "CREATE TABLE "+this.getStoreTableName()+
                 " (" +
                 "id BIGSERIAL PRIMARY KEY,"+
-                COMMAND_ID + " bigint, "+
-                "data jsonb"+
+                COMMAND_ID+" BIGINT, "+
+                PROGRESS+" INT, "+
+                REPORT_TIME+" TIMESTAMP, "+
+                STATUS_CODE+" VARCHAR, "+
+                EXEC_TIME+" tsrange, "+
+                MESSAGE+" VARCHAR, "+
+                RESULT+" json"+
                 ")";
     }
 
     public String insertCommandQuery() {
-        return "INSERT INTO "+this.getStoreTableName()+" (commandid, data) VALUES (?, ?)";
+        return "INSERT INTO "+this.getStoreTableName()+" " +
+                "(commandid, progress, reportTime, statusCode, executionTime, message, result) " +
+                "VALUES (?, ?, ?, ?, ?, ?, ?)";
     }
 
     public String updateByIdQuery() {
-        return "UPDATE "+this.getStoreTableName()+" SET data = ?, WHERE id = ?";
+        return "UPDATE "+this.getStoreTableName()+" SET "+
+                COMMAND_ID+" = ?, " +
+                PROGRESS+" = ?, " +
+                REPORT_TIME+" = ?, " +
+                STATUS_CODE+" = ?, " +
+                EXEC_TIME+" = ?, " +
+                MESSAGE+" = ?, " +
+                RESULT+" = ?, " +
+                "WHERE id = ?";
     }
 
     public String createSelectEntriesQuery(CommandStatusFilter filter, Set<ICommandStatusStore.CommandStatusField> fields) {

persistence/sensorhub-datastore-postgis/src/main/java/org/sensorhub/impl/datastore/postgis/builder/QueryBuilderCommandStore.java

@@ -42,16 +42,16 @@ public String createTableQuery() {
                 COMMANDSTREAM_ID +" BIGINT, "+
                 SENDER_ID+" VARCHAR,"+
                 FOI_ID+" bigint,"+
-                ISSUE_TIME+" TIMESTAMPTZ,"+
+                ISSUE_TIME+" TIMESTAMP,"+
                 STATUS+" VARCHAR,"+
                 ERROR_MSG+" VARCHAR,"+
-                PARAMETERS+" JSONB" +
+                PARAMETERS+" JSON" +
                 ")";
     }
 
-    public String createDataIndexQuery() {
-        return "CREATE INDEX "+this.getStoreTableName()+"_data_idx on "+this.getStoreTableName()+" USING GIN("+PARAMETERS+")";
-    }
+//    public String createDataIndexQuery() {
+//        return "CREATE INDEX "+this.getStoreTableName()+"_data_idx on "+this.getStoreTableName()+" ("+PARAMETERS+")";
+//    }
 
     public String createCommandStreamIndexQuery() {
         return "CREATE INDEX "+this.getStoreTableName()+"_commandstream_idx on "+this.getStoreTableName()+" ("+COMMANDSTREAM_ID+")";

persistence/sensorhub-datastore-postgis/src/main/java/org/sensorhub/impl/datastore/postgis/builder/QueryBuilderCommandStreamStore.java

@@ -21,7 +21,7 @@
 import java.util.Set;
 
 public class QueryBuilderCommandStreamStore extends QueryBuilder {
-    public final static String COMMAND_STREAM_TABLE_NAME = "commandstream";
+    public final static String COMMAND_STREAM_TABLE_NAME = "commandstreams";
 
     public QueryBuilderCommandStreamStore() {
         this(COMMAND_STREAM_TABLE_NAME);

persistence/sensorhub-datastore-postgis/src/main/java/org/sensorhub/impl/datastore/postgis/builder/QueryBuilderDataStreamStore.java

@@ -89,8 +89,8 @@ public String insertInfoQuery() {
     }
 
     public String getAllDataStreams(Instant min, Instant max) {
-        return "SELECT id FROM "+this.getStoreTableName()+" WHERE tstzrange((data->'validTime'->>'begin')::timestamptz," +
-                "(data->'validTime'->>'end')::timestamptz) <@ '["+ PostgisUtils.checkAndGetValidInstant(min)+","+PostgisUtils.checkAndGetValidInstant(max)+"]'::tstzrange";
+        return "SELECT id FROM "+this.getStoreTableName()+" WHERE tsrange((data->'validTime'->>'begin')::timestamp," +
+                "(data->'validTime'->>'end')::timestamp) <@ '["+ PostgisUtils.checkAndGetValidInstant(min)+","+PostgisUtils.checkAndGetValidInstant(max)+"]'::tsrange";
     }
 
     public String createSelectEntriesQuery(DataStreamFilter filter, Set<IDataStreamStore.DataStreamInfoField> fields) {

persistence/sensorhub-datastore-postgis/src/main/java/org/sensorhub/impl/datastore/postgis/builder/QueryBuilderObsStore.java

@@ -39,35 +39,34 @@ public QueryBuilderObsStore(String tableName) {
     }
 
     public String createTableQuery() {
-        return "CREATE TABLE "+this.getStoreTableName()+
+        return "CREATE TABLE IF NOT EXISTS "+this.getStoreTableName()+
                 " (" +
                 "id BIGSERIAL PRIMARY KEY,"+
                 DATASTREAM_ID +" bigint, "+
                 FOI_ID+" bigint,"+
-                PHENOMENON_TIME+" TIMESTAMPTZ,"+
-                RESULT_TIME+" TIMESTAMPTZ,"+
-                RESULT+" JSON" + // VERSUS JSONB but the parser does not keep order
+                PHENOMENON_TIME+" TIMESTAMP,"+
+                RESULT_TIME+" TIMESTAMP,"+
+                RESULT+" JSONB" + // VERSUS JSONB but the parser does not keep order
                 ")";
     }
     public String createDataIndexQuery() {
         return "CREATE INDEX "+this.getStoreTableName()+"_data_idx on "+this.getStoreTableName()+" USING GIN("+RESULT+")";
-//        return "CREATE INDEX "+this.getStoreTableName()+"_data_idx on "+this.getStoreTableName()+" ("+RESULT+")";
     }
 
     public String createDataStreamIndexQuery() {
-        return "CREATE INDEX "+this.getStoreTableName()+"_datastream_idx on "+this.getStoreTableName()+" ("+DATASTREAM_ID+")";
+        return "CREATE INDEX IF NOT EXISTS "+this.getStoreTableName()+"_datastream_idx on "+this.getStoreTableName()+" ("+DATASTREAM_ID+")";
     }
 
     public String createPhenomenonTimeIndexQuery() {
-        return "CREATE INDEX "+this.getStoreTableName()+"_phenomenon_time_idx on "+this.getStoreTableName()+" ("+ PHENOMENON_TIME +")";
+        return "CREATE INDEX IF NOT EXISTS "+this.getStoreTableName()+"_datastream_id_phenomenon_time_idx on "+this.getStoreTableName()+" ("+ DATASTREAM_ID + ", " + PHENOMENON_TIME +")";
     }
 
     public String createResultTimeIndexQuery() {
-        return "CREATE INDEX "+this.getStoreTableName()+"_result_time_idx on "+this.getStoreTableName()+" ("+ RESULT_TIME +")";
+        return "CREATE INDEX IF NOT EXISTS "+this.getStoreTableName()+"_datastream_id_result_time_idx on "+this.getStoreTableName()+" ("+ DATASTREAM_ID + ", " + RESULT_TIME +")";
     }
 
     public String createFoiIndexQuery() {
-        return "CREATE INDEX "+this.getStoreTableName()+"_foi_idx on "+this.getStoreTableName()+" ("+ FOI_ID +")";
+        return "CREATE INDEX IF NOT EXISTS "+this.getStoreTableName()+"_foi_idx on "+this.getStoreTableName()+" ("+ FOI_ID +")";
     }
 
     public String insertObsQuery() {
@@ -77,7 +76,7 @@ public String insertObsQuery() {
     }
 
     public String createUniqueConstraint() {
-        return "CREATE UNIQUE INDEX  "+this.getStoreTableName()+"_unique_constraint on "+this.getStoreTableName()+" (dataStreamID, foiID, phenomenonTime, resultTime)";
+        return "CREATE UNIQUE INDEX  IF NOT EXISTS "+this.getStoreTableName()+"_unique_constraint on "+this.getStoreTableName()+" (dataStreamID, foiID, phenomenonTime, resultTime)";
     }
 
     public String updateByIdQuery() {
@@ -93,6 +92,15 @@ public String getPhenomenonTimeRangeByDataStreamIdQuery(long dataStreamID) {
         return "SELECT Min("+PHENOMENON_TIME+"),Max("+PHENOMENON_TIME+") FROM "+this.getStoreTableName()+" WHERE "+DATASTREAM_ID+" = "+dataStreamID;
     }
 
+    public String getPhenomenonTimeRangeByDataStreamIdsQuery() {
+        return "SELECT " + DATASTREAM_ID + ", " +
+                "       MIN(" + PHENOMENON_TIME + ") AS min, " +
+                "       MAX(" + PHENOMENON_TIME + ") AS max " +
+                "FROM " + getStoreTableName() +
+                " WHERE " + DATASTREAM_ID + " = ANY (?) " +
+                "GROUP BY " + DATASTREAM_ID;
+    }
+
     public String getBinCountByPhenomenontime(long seconds, List<Long> dsIds, List<Long> foiIds) {
         StringBuilder idsQuery = new StringBuilder();
         for(int i=0;i < dsIds.size();i++) {
@@ -118,6 +126,15 @@ public String getResultTimeRangeByDataStreamIdQuery(long dataStreamID) {
         return "SELECT Min("+RESULT_TIME+"),Max("+RESULT_TIME+") FROM "+this.getStoreTableName()+" WHERE "+DATASTREAM_ID+" = "+dataStreamID;
     }
 
+    public String getResultTimeRangeByDataStreamIdsQuery() {
+        return "SELECT " + DATASTREAM_ID + ", " +
+                "       MIN(" + RESULT_TIME + ") AS min, " +
+                "       MAX(" + RESULT_TIME + ") AS max " +
+                "FROM " + getStoreTableName() +
+                " WHERE " + DATASTREAM_ID + " = ANY (?) " +
+                "GROUP BY " + DATASTREAM_ID;
+    }
+
     public String countByPhenomenonTimeRangeQuery(Instant min, Instant max) {
             String minTimestamp = min.toString();
             if(min == Instant.MIN) {
@@ -127,8 +144,8 @@ public String countByPhenomenonTimeRangeQuery(Instant min, Instant max) {
             if(max == Instant.MAX) {
                 maxTimestamp = "infinity";
             }
-            String sb = "tstzrange((" + this.getStoreTableName() + "."+PHENOMENON_TIME+")::timestamptz," +
-                    " (" + this.getStoreTableName()  + "."+PHENOMENON_TIME+")::timestamptz)" +
+            String sb = "tsrange((" + this.getStoreTableName() + "."+PHENOMENON_TIME+")::timestamp," +
+                    " (" + this.getStoreTableName()  + "."+PHENOMENON_TIME+")::timestamp)" +
                     " && '[" + minTimestamp + "," + maxTimestamp + "]'";
 
         return "SELECT COUNT(*) FROM "+this.getStoreTableName()+" "+sb;