[influxdb] Move keep() after ordering/pagination for query performance

Applying keep() before last()/sort()/limit() prevents InfluxDB from using
time-series index pushdown optimisations. For the common restore-on-startup
case (DESCENDING + pageSize=1), keep() → last() forces a full data scan
before last() can run; reversing the order to last() → keep() allows the
query planner to seek directly to the most recent record.

The reordering also fixes a latent correctness issue: the state-value filter
uses r._field, which keep() was dropping before the filter ran.

Signed-off-by: Jonathan Gilbert <jpg@trillica.com>

Author: jpg0

bundles/org.openhab.persistence.influxdb/src/main/java/org/openhab/persistence/influxdb/internal/influx2/InfluxDB2FilterCriteriaQueryCreatorImpl.java

@@ -69,12 +69,9 @@ public String createQuery(FilterCriteria criteria, String retentionPolicy, @Null
         String name = influxDBMetadataService.getMeasurementNameOrDefault(localAlias);
         String measurementName = configuration.isReplaceUnderscore() ? name.replace('_', '.') : name;
         flux = flux.filter(measurement().equal(measurementName));
-        if (!measurementName.equals(itemName)) {
+        boolean needsItemTag = !measurementName.equals(itemName);
+        if (needsItemTag) {
             flux = flux.filter(tag(TAG_ITEM_NAME).equal(itemName));
-            flux = flux.keep(
-                    new String[] { FIELD_MEASUREMENT_NAME, COLUMN_TIME_NAME_V2, COLUMN_VALUE_NAME_V2, TAG_ITEM_NAME });
-        } else {
-            flux = flux.keep(new String[] { FIELD_MEASUREMENT_NAME, COLUMN_TIME_NAME_V2, COLUMN_VALUE_NAME_V2 });
         }
 
         State filterState = criteria.getState();
@@ -84,8 +81,18 @@ public String createQuery(FilterCriteria criteria, String retentionPolicy, @Null
             flux = flux.filter(restrictions);
         }
 
+        // Apply ordering/pagination before keep() so that last()/sort()/limit() can benefit
+        // from InfluxDB's pushdown optimisations. Applying keep() first drops _field, which
+        // both prevents state-value filtering from working correctly and blocks last() pushdown.
         flux = applyOrderingAndPageSize(criteria, flux);
 
+        if (needsItemTag) {
+            flux = flux.keep(
+                    new String[] { FIELD_MEASUREMENT_NAME, COLUMN_TIME_NAME_V2, COLUMN_VALUE_NAME_V2, TAG_ITEM_NAME });
+        } else {
+            flux = flux.keep(new String[] { FIELD_MEASUREMENT_NAME, COLUMN_TIME_NAME_V2, COLUMN_VALUE_NAME_V2 });
+        }
+
         return flux.toString();
     }
 

bundles/org.openhab.persistence.influxdb/src/test/java/org/openhab/persistence/influxdb/internal/InfluxFilterCriteriaQueryCreatorImplTest.java

@@ -85,8 +85,8 @@ public void testSimpleItemQueryWithoutParams() {
                 from(bucket:"origin")
                 \t|> range(start:-100y, stop:100y)
                 \t|> filter(fn: (r) => r["_measurement"] == "sampleItem")
-                \t|> keep(columns:["_measurement", "_time", "_value"])
-                \t|> sort(desc:true, columns:["_time"])"""));
+                \t|> sort(desc:true, columns:["_time"])
+                \t|> keep(columns:["_measurement", "_time", "_value"])"""));
     }
 
     @Test
@@ -108,9 +108,9 @@ public void testRangeCriteria() {
                 from(bucket:"origin")
                 \t|> range(start:%s, stop:%s)
                 \t|> filter(fn: (r) => r["_measurement"] == "sampleItem")
-                \t|> keep(columns:["_measurement", "_time", "_value"])
-                \t|> sort(desc:true, columns:["_time"])""", INFLUX2_DATE_FORMATTER.format(now.toInstant()),
-                INFLUX2_DATE_FORMATTER.format(tomorrow.toInstant()));
+                \t|> sort(desc:true, columns:["_time"])
+                \t|> keep(columns:["_measurement", "_time", "_value"])""",
+                INFLUX2_DATE_FORMATTER.format(now.toInstant()), INFLUX2_DATE_FORMATTER.format(tomorrow.toInstant()));
         assertThat(queryV2, equalTo(expectedQueryV2));
     }
 
@@ -129,9 +129,9 @@ public void testValueOperator() {
                 from(bucket:"origin")
                 \t|> range(start:-100y, stop:100y)
                 \t|> filter(fn: (r) => r["_measurement"] == "sampleItem")
-                \t|> keep(columns:["_measurement", "_time", "_value"])
                 \t|> filter(fn: (r) => (r["_field"] == "value" and r["_value"] <= 90))
-                \t|> sort(desc:true, columns:["_time"])"""));
+                \t|> sort(desc:true, columns:["_time"])
+                \t|> keep(columns:["_measurement", "_time", "_value"])"""));
     }
 
     @Test
@@ -149,9 +149,9 @@ public void testPagination() {
                 from(bucket:"origin")
                 \t|> range(start:-100y, stop:100y)
                 \t|> filter(fn: (r) => r["_measurement"] == "sampleItem")
-                \t|> keep(columns:["_measurement", "_time", "_value"])
                 \t|> sort(desc:true, columns:["_time"])
-                \t|> limit(n:10, offset:20)"""));
+                \t|> limit(n:10, offset:20)
+                \t|> keep(columns:["_measurement", "_time", "_value"])"""));
     }
 
     @Test
@@ -168,8 +168,8 @@ public void testOrdering() {
                 from(bucket:"origin")
                 \t|> range(start:-100y, stop:100y)
                 \t|> filter(fn: (r) => r["_measurement"] == "sampleItem")
-                \t|> keep(columns:["_measurement", "_time", "_value"])
-                \t|> sort(desc:false, columns:["_time"])"""));
+                \t|> sort(desc:false, columns:["_time"])
+                \t|> keep(columns:["_measurement", "_time", "_value"])"""));
     }
 
     @Test
@@ -182,8 +182,8 @@ public void testPreviousState() {
                 from(bucket:"origin")
                 \t|> range(start:-100y, stop:100y)
                 \t|> filter(fn: (r) => r["_measurement"] == "sampleItem")
-                \t|> keep(columns:["_measurement", "_time", "_value"])
-                \t|> last()"""));
+                \t|> last()
+                \t|> keep(columns:["_measurement", "_time", "_value"])"""));
     }
 
     private FilterCriteria createBaseCriteria() {
@@ -210,8 +210,8 @@ public void testMeasurementNameFromMetadata() {
                 \t|> range(start:-100y, stop:100y)
                 \t|> filter(fn: (r) => r["_measurement"] == "measurementName")
                 \t|> filter(fn: (r) => r["item"] == "sampleItem")
-                \t|> keep(columns:["_measurement", "_time", "_value", "item"])
-                \t|> sort(desc:true, columns:["_time"])"""));
+                \t|> sort(desc:true, columns:["_time"])
+                \t|> keep(columns:["_measurement", "_time", "_value", "item"])"""));
         when(metadataRegistry.get(metadataKey))
                 .thenReturn(new Metadata(metadataKey, "", Map.of("key1", "val1", "key2", "val2")));
 
@@ -224,7 +224,7 @@ public void testMeasurementNameFromMetadata() {
                 from(bucket:"origin")
                 \t|> range(start:-100y, stop:100y)
                 \t|> filter(fn: (r) => r["_measurement"] == "sampleItem")
-                \t|> keep(columns:["_measurement", "_time", "_value"])
-                \t|> sort(desc:true, columns:["_time"])"""));
+                \t|> sort(desc:true, columns:["_time"])
+                \t|> keep(columns:["_measurement", "_time", "_value"])"""));
     }
 }