Feat/consys custom formats

sensorhub-core/src/main/java/org/sensorhub/api/datastore/command/ICommandStreamStore.java

@@ -14,6 +14,8 @@
 
 package org.sensorhub.api.datastore.command;
 
+import java.time.Instant;
+import java.util.Comparator;
 import org.sensorhub.api.command.ICommandStreamInfo;
 import org.sensorhub.api.datastore.DataStoreException;
 import org.sensorhub.api.datastore.command.ICommandStreamStore.CommandStreamInfoField;
@@ -111,12 +113,20 @@ public default Entry<CommandStreamKey, ICommandStreamInfo> getLatestVersionEntry
                 .done()
             .withControlInputNames(controlInput)
             .build())
-        .findFirst();
+        // results are not guaranteed to be sorted, so select the version with the latest valid time
+        .max(Comparator.comparing(e -> getValidTimeBegin(e.getValue())));
 
         return entryOpt.isPresent() ? entryOpt.get() : null;
     }
 
 
+    private static Instant getValidTimeBegin(ICommandStreamInfo info)
+    {
+        var validTime = info.getValidTime();
+        return validTime != null ? validTime.begin() : Instant.MIN;
+    }
+
+
     /**
      * Remove all command streams that are associated to the given system command input
      * @param sysUID Unique ID of system receiving the command stream

sensorhub-core/src/main/java/org/sensorhub/api/datastore/obs/IDataStreamStore.java

@@ -14,6 +14,8 @@
 
 package org.sensorhub.api.datastore.obs;
 
+import java.time.Instant;
+import java.util.Comparator;
 import org.sensorhub.api.data.IDataStreamInfo;
 import org.sensorhub.api.datastore.DataStoreException;
 import org.sensorhub.api.datastore.obs.IDataStreamStore.DataStreamInfoField;
@@ -111,12 +113,20 @@ public default Entry<DataStreamKey, IDataStreamInfo> getLatestVersionEntry(Strin
                 .done()
             .withOutputNames(outputName)
             .build())
-        .findFirst();
+        // results are not guaranteed to be sorted, so select the version with the latest valid time
+        .max(Comparator.comparing(e -> getValidTimeBegin(e.getValue())));
 
         return entryOpt.isPresent() ? entryOpt.get() : null;
     }
 
 
+    private static Instant getValidTimeBegin(IDataStreamInfo info)
+    {
+        var validTime = info.getValidTime();
+        return validTime != null ? validTime.begin() : Instant.MIN;
+    }
+
+
     /**
      * Remove all datastreams that are associated to the given system output
      * @param sysUID

sensorhub-core/src/test/java/org/sensorhub/impl/datastore/mem/TestInMemCommandStreamStore.java

@@ -14,7 +14,24 @@
 
 package org.sensorhub.impl.datastore.mem;
 
+import static org.junit.Assert.assertEquals;
+import java.time.Instant;
+import java.util.AbstractMap.SimpleEntry;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.stream.Stream;
+import org.junit.Test;
+import org.sensorhub.api.command.CommandStreamInfo;
+import org.sensorhub.api.command.ICommandStreamInfo;
+import org.sensorhub.api.common.BigId;
+import org.sensorhub.api.datastore.command.CommandStreamFilter;
+import org.sensorhub.api.datastore.command.CommandStreamKey;
+import org.sensorhub.api.datastore.command.ICommandStreamStore.CommandStreamInfoField;
+import org.sensorhub.api.feature.FeatureId;
 import org.sensorhub.impl.datastore.AbstractTestCommandStreamStore;
+import org.vast.data.TextEncodingImpl;
+import org.vast.swe.SWEHelper;
+import org.vast.util.TimeExtent;
 
 
 public class TestInMemCommandStreamStore extends AbstractTestCommandStreamStore<InMemoryCommandStreamStore>
@@ -73,4 +90,34 @@ protected void testAddAndSelectByKeywords_ExpectedResults(int testCaseIdx)
             case 4: addToExpectedResults(4); break;
         }        
     }
+
+
+    @Test
+    public void testLatestVersionEntryWithUnsortedResults() throws Exception
+    {
+        var sysId = new FeatureId(BigId.fromLong(DATABASE_NUM, 1), "urn:osh:test:sys1");
+        var rec = new SWEHelper().createRecord().name("in1").addField("f1", new SWEHelper().createQuantity()).build();
+        var oldInfo = new CommandStreamInfo.Builder()
+            .withName("v1").withSystem(sysId).withRecordDescription(rec).withRecordEncoding(new TextEncodingImpl())
+            .withValidTime(TimeExtent.beginAt(Instant.parse("2020-01-01T00:00:00Z")))
+            .build();
+        var newInfo = new CommandStreamInfo.Builder()
+            .withName("v2").withSystem(sysId).withRecordDescription(rec).withRecordEncoding(new TextEncodingImpl())
+            .withValidTime(TimeExtent.beginAt(Instant.parse("2024-01-01T00:00:00Z")))
+            .build();
+
+        // store that returns versions oldest first
+        var store = new InMemoryCommandStreamStore(new InMemoryCommandStore(DATABASE_NUM)) {
+            @Override
+            public Stream<Entry<CommandStreamKey, ICommandStreamInfo>> selectEntries(CommandStreamFilter filter, Set<CommandStreamInfoField> fields)
+            {
+                return Stream.of(
+                    new SimpleEntry<>(new CommandStreamKey(DATABASE_NUM, 1), oldInfo),
+                    new SimpleEntry<>(new CommandStreamKey(DATABASE_NUM, 2), newInfo));
+            }
+        };
+
+        var e = store.getLatestVersionEntry(sysId.getUniqueID(), "in1");
+        assertEquals("v2", e.getValue().getName());
+    }
 }

sensorhub-core/src/test/java/org/sensorhub/impl/datastore/mem/TestInMemDataStreamStore.java

@@ -14,7 +14,24 @@
 
 package org.sensorhub.impl.datastore.mem;
 
+import static org.junit.Assert.assertEquals;
+import java.time.Instant;
+import java.util.AbstractMap.SimpleEntry;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.stream.Stream;
+import org.junit.Test;
+import org.sensorhub.api.common.BigId;
+import org.sensorhub.api.data.DataStreamInfo;
+import org.sensorhub.api.data.IDataStreamInfo;
+import org.sensorhub.api.datastore.obs.DataStreamFilter;
+import org.sensorhub.api.datastore.obs.DataStreamKey;
+import org.sensorhub.api.datastore.obs.IDataStreamStore.DataStreamInfoField;
+import org.sensorhub.api.feature.FeatureId;
 import org.sensorhub.impl.datastore.AbstractTestDataStreamStore;
+import org.vast.data.TextEncodingImpl;
+import org.vast.swe.SWEHelper;
+import org.vast.util.TimeExtent;
 
 
 public class TestInMemDataStreamStore extends AbstractTestDataStreamStore<InMemoryDataStreamStore>
@@ -73,4 +90,34 @@ protected void testAddAndSelectByKeywords_ExpectedResults(int testCaseIdx)
             case 4: addToExpectedResults(4); break;
         }        
     }
+
+
+    @Test
+    public void testLatestVersionEntryWithUnsortedResults() throws Exception
+    {
+        var sysId = new FeatureId(BigId.fromLong(DATABASE_NUM, 1), "urn:osh:test:sys1");
+        var rec = new SWEHelper().createRecord().name("out1").addField("f1", new SWEHelper().createQuantity()).build();
+        var oldInfo = new DataStreamInfo.Builder()
+            .withName("v1").withSystem(sysId).withRecordDescription(rec).withRecordEncoding(new TextEncodingImpl())
+            .withValidTime(TimeExtent.beginAt(Instant.parse("2020-01-01T00:00:00Z")))
+            .build();
+        var newInfo = new DataStreamInfo.Builder()
+            .withName("v2").withSystem(sysId).withRecordDescription(rec).withRecordEncoding(new TextEncodingImpl())
+            .withValidTime(TimeExtent.beginAt(Instant.parse("2024-01-01T00:00:00Z")))
+            .build();
+
+        // store that returns versions oldest first
+        var store = new InMemoryDataStreamStore(new InMemoryObsStore(DATABASE_NUM)) {
+            @Override
+            public Stream<Entry<DataStreamKey, IDataStreamInfo>> selectEntries(DataStreamFilter filter, Set<DataStreamInfoField> fields)
+            {
+                return Stream.of(
+                    new SimpleEntry<>(new DataStreamKey(DATABASE_NUM, 1), oldInfo),
+                    new SimpleEntry<>(new DataStreamKey(DATABASE_NUM, 2), newInfo));
+            }
+        };
+
+        var e = store.getLatestVersionEntry(sysId.getUniqueID(), "out1");
+        assertEquals("v2", e.getValue().getName());
+    }
 }

sensorhub-core/src/testFixtures/java/org/sensorhub/impl/datastore/AbstractTestCommandStore.java

@@ -483,6 +483,39 @@ public void testSelectCommandsByMultipleIDs() throws Exception
     }
 
 
+    @Test
+    public void testRemoveByKeyDoesNotMixCommandStreams() throws Exception
+    {
+        var cs1ID = addSimpleCommandStream(bigId(10), "out1").getInternalID();
+        var cs2ID = addSimpleCommandStream(bigId(20), "out2").getInternalID();
+
+        // empty the newest command series by removing its only command
+        var removed = addCommands(cs1ID, Instant.parse("2020-01-01T00:00:00Z"), 1, 1000);
+        for (var k: removed.keySet())
+        {
+            cmdStore.remove(k);
+            allCommands.remove(k);
+        }
+        forceReadBackFromStorage();
+
+        // new series for command stream 2, then more commands in the existing series of stream 1
+        var cs2Cmds = addCommands(cs2ID, Instant.parse("2021-01-01T00:00:00Z"), 10, 1000);
+        forceReadBackFromStorage();
+        var cs1Cmds = addCommands(cs1ID, Instant.parse("2022-01-01T00:00:00Z"), 3, 1000);
+        forceReadBackFromStorage();
+
+        var filter = new CommandFilter.Builder()
+            .withCommandStreams(cs1ID)
+            .build();
+        checkSelectedEntries(cmdStore.selectEntries(filter), cs1Cmds, filter);
+
+        filter = new CommandFilter.Builder()
+            .withCommandStreams(cs2ID)
+            .build();
+        checkSelectedEntries(cmdStore.selectEntries(filter), cs2Cmds, filter);
+    }
+
+
     @Test
     public void testSelectCommandsByDataStreamIDAndTime() throws Exception
     {

sensorhub-core/src/testFixtures/java/org/sensorhub/impl/datastore/AbstractTestObsStore.java

@@ -383,6 +383,66 @@ public void testAddAndRemoveByKey() throws Exception
     }
 
 
+    @Test
+    public void testRemoveByKeyDoesNotMixDataStreams() throws Exception
+    {
+        var ds1ID = addSimpleDataStream(bigId(10), "out1");
+        var ds2ID = addSimpleDataStream(bigId(20), "out2");
+
+        // empty the newest obs series by removing its only obs
+        var removed = addSimpleObsWithoutResultTime(ds1ID, BigId.NONE, Instant.parse("2020-01-01T00:00:00Z"), 1);
+        for (var k: removed.keySet())
+        {
+            obsStore.remove(k);
+            allObs.remove(k);
+        }
+        forceReadBackFromStorage();
+
+        // new series for datastream 2, then more obs in the existing series of datastream 1
+        var ds2Obs = addSimpleObsWithoutResultTime(ds2ID, BigId.NONE, Instant.parse("2021-01-01T00:00:00Z"), 10);
+        forceReadBackFromStorage();
+        var ds1Obs = addSimpleObsWithoutResultTime(ds1ID, BigId.NONE, Instant.parse("2022-01-01T00:00:00Z"), 3);
+        forceReadBackFromStorage();
+
+        var filter = new ObsFilter.Builder()
+            .withDataStreams(ds1ID)
+            .build();
+        checkSelectedEntries(obsStore.selectEntries(filter), ds1Obs, filter);
+
+        filter = new ObsFilter.Builder()
+            .withDataStreams(ds2ID)
+            .build();
+        checkSelectedEntries(obsStore.selectEntries(filter), ds2Obs, filter);
+    }
+
+
+    @Test
+    public void testSelectLatestResultWithinPhenomenonTimeRange() throws Exception
+    {
+        var dsID = addSimpleDataStream(bigId(10), "out1");
+        var startTime = Instant.parse("2020-01-01T00:00:00Z");
+        var obs = new ArrayList<>(addSimpleObsWithoutResultTime(dsID, BigId.NONE, startTime, 5, 60000).entrySet());
+        forceReadBackFromStorage();
+
+        // range covers the 2nd to 3rd obs, so only the 3rd is expected
+        var filter = new ObsFilter.Builder()
+            .withDataStreams(dsID)
+            .withPhenomenonTimeDuring(startTime.plusSeconds(60), startTime.plusSeconds(150))
+            .withLatestResult()
+            .build();
+        var expected = Map.of(obs.get(2).getKey(), obs.get(2).getValue());
+        checkSelectedEntries(obsStore.selectEntries(filter), expected, filter);
+
+        // range after all obs
+        filter = new ObsFilter.Builder()
+            .withDataStreams(dsID)
+            .withPhenomenonTimeDuring(startTime.plusSeconds(3600), startTime.plusSeconds(7200))
+            .withLatestResult()
+            .build();
+        checkSelectedEntries(obsStore.selectEntries(filter), Collections.emptyMap(), filter);
+    }
+
+
     @Test
     public void testAddAndRemoveEntireDatastream() throws Exception
     {

sensorhub-datastore-h2/src/main/java/org/sensorhub/impl/datastore/h2/MVCommandStoreImpl.java

@@ -66,6 +66,7 @@ public class MVCommandStoreImpl implements ICommandStore
     protected MVCommandStatusStoreImpl cmdStatusStore;
     protected MVBTreeMap<MVTimeSeriesRecordKey, ICommandData> cmdRecordsIndex;
     protected MVBTreeMap<MVTimeSeriesKey, MVTimeSeriesInfo> cmdSeriesMainIndex;
+    protected long lastSeriesID;
     protected int idScope;
 
     protected IFoiStore foiStore;
@@ -143,6 +144,12 @@ private MVCommandStoreImpl init(MVStore mvStore, int idScope, IdProviderType dsI
                 .keyType(new MVCommandSeriesKeyByCommandStreamDataType())
                 .valueType(new MVTimeSeriesInfoDataType()));
 
+        // next series ID must follow every series still registered, including empty ones,
+        // otherwise a new series can reuse the ID of an existing one
+        this.lastSeriesID = Math.max(
+            cmdSeriesMainIndex.values().stream().mapToLong(s -> s.id).max().orElse(0),
+            cmdRecordsIndex.isEmpty() ? 0 : cmdRecordsIndex.lastKey().seriesID);
+
         return this;
     }
 
@@ -621,7 +628,7 @@ public synchronized BigId add(ICommandData cmd)
 
         MVTimeSeriesInfo series = cmdSeriesMainIndex.computeIfAbsent(seriesKey, k -> {
             return new MVTimeSeriesInfo(
-                cmdRecordsIndex.isEmpty() ? 1 : cmdRecordsIndex.lastKey().seriesID + 1);
+                ++lastSeriesID);
         });
 
         // add to main command index

sensorhub-datastore-h2/src/main/java/org/sensorhub/impl/datastore/h2/MVObsStoreImpl.java

@@ -74,6 +74,7 @@ public class MVObsStoreImpl implements IObsStore
     protected MVDataStreamStoreImpl dataStreamStore;
     protected MVBTreeMap<MVTimeSeriesRecordKey, IObsData> obsRecordsIndex;
     protected MVBTreeMap<MVTimeSeriesKey, MVTimeSeriesInfo> obsSeriesMainIndex;
+    protected long lastSeriesID;
     protected MVBTreeMap<MVTimeSeriesKey, Boolean> obsSeriesByFoiIndex;
     protected int idScope;
 
@@ -165,6 +166,12 @@ private MVObsStoreImpl init(MVStore mvStore, int idScope, IdProviderType dsIdPro
                 .keyType(new MVObsSeriesKeyByFoiDataType())
                 .valueType(new MVVoidDataType()));
 
+        // next series ID must follow every series still registered, including empty ones,
+        // otherwise a new series can reuse the ID of an existing one
+        this.lastSeriesID = Math.max(
+            obsSeriesMainIndex.values().stream().mapToLong(s -> s.id).max().orElse(0),
+            obsRecordsIndex.isEmpty() ? 0 : obsRecordsIndex.lastKey().seriesID);
+
         return this;
     }
 
@@ -303,11 +310,12 @@ Stream<Entry<MVTimeSeriesRecordKey, IObsData>> getObsStream(MVTimeSeriesInfo ser
             }
 
             // if request if for latest result only, get only the latest obs in series
+            // within the requested phenomenon time range
             if (timeParams.latestResultOnly)
             {
-                MVTimeSeriesRecordKey maxKey = new MVTimeSeriesRecordKey(series.id, Instant.MAX);
+                MVTimeSeriesRecordKey maxKey = new MVTimeSeriesRecordKey(series.id, phenomenonTimeRange.upperEndpoint());
                 Entry<MVTimeSeriesRecordKey, IObsData> e = obsRecordsIndex.floorEntry(maxKey);
-                if (e != null && e.getKey().seriesID == series.id)
+                if (e != null && e.getKey().seriesID == series.id && phenomenonTimeRange.contains(e.getKey().timeStamp))
                     return Stream.of(e);
                 else
                     return Stream.empty();
@@ -908,7 +916,7 @@ public synchronized BigId add(IObsData obs)
             obsSeriesByFoiIndex.putIfAbsent(seriesKey, Boolean.TRUE);
 
             return new MVTimeSeriesInfo(
-                obsRecordsIndex.isEmpty() ? 1 : obsRecordsIndex.lastKey().seriesID + 1);
+                ++lastSeriesID);
         });
 
         // add to main obs index