Verify step is aligned with range selector lookback

Author: felixbarny

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/logical/promql/PromqlCommand.java

@@ -13,8 +13,10 @@
 import org.elasticsearch.xpack.esql.common.Failures;
 import org.elasticsearch.xpack.esql.core.tree.NodeInfo;
 import org.elasticsearch.xpack.esql.core.tree.Source;
+import org.elasticsearch.xpack.esql.expression.promql.subquery.Subquery;
 import org.elasticsearch.xpack.esql.plan.logical.LogicalPlan;
 import org.elasticsearch.xpack.esql.plan.logical.UnaryPlan;
+import org.elasticsearch.xpack.esql.plan.logical.promql.selector.RangeSelector;
 import org.elasticsearch.xpack.esql.plan.logical.promql.selector.Selector;
 
 import java.io.IOException;
@@ -140,12 +142,27 @@ public void postAnalysisVerification(Failures failures) {
         if (p instanceof AcrossSeriesAggregate == false) {
             failures.add(fail(p, "only aggregations across timeseries are supported at this time (found [{}])", p.sourceText()));
         }
-        p.forEachDown(Selector.class, s -> {
-            if (s.labelMatchers().nameLabel().matcher().isRegex()) {
-                failures.add(fail(s, "regex label selectors on __name__ are not supported at this time [{}]", s.sourceText()));
+        p.forEachDown(lp -> {
+            if (lp instanceof Selector s) {
+                if (s.labelMatchers().nameLabel().matcher().isRegex()) {
+                    failures.add(fail(s, "regex label selectors on __name__ are not supported at this time [{}]", s.sourceText()));
+                }
+                if (s.evaluation() != null && s.evaluation().offset() != null && s.evaluation().offset().isZero() == false) {
+                    failures.add(fail(s, "offset modifiers are not supported at this time [{}]", s.sourceText()));
+                }
             }
-            if (s.evaluation() != null && s.evaluation().offset() != null && s.evaluation().offset().isZero() == false) {
-                failures.add(fail(s, "offset modifiers are not supported at this time [{}]", s.sourceText()));
+            if (step() != null && lp instanceof RangeSelector rs) {
+                Duration rangeDuration = (Duration) rs.range().fold(null);
+                if (rangeDuration.equals(step()) == false) {
+                    failures.add(
+                        fail(
+                            rs.range(),
+                            "the duration for range vector selector [{}] "
+                                + "must be equal to the query's step for range queries at this time",
+                            rs.range().sourceText()
+                        )
+                    );
+                }
             }
         });
     }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/analysis/promql/PromqlVerifierTests.java

@@ -32,6 +32,16 @@ public void testPromqlMissingAcrossSeriesAggregation() {
         );
     }
 
+    public void testPromqlStepAndRangeMisaligned() {
+        assertThat(
+            error("""
+                TS test | PROMQL step 1m (
+                  avg(rate(network.bytes_in[5m]))
+                )""", tsdb),
+            equalTo("2:29: the duration for range vector selector [5m] must be equal to the query's step for range queries at this time")
+        );
+    }
+
     public void testPromqlIllegalNameLabelMatcher() {
         assertThat(
             error("TS test | PROMQL step 5m (avg({__name__=~\"*.foo.*\"}))", tsdb),

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/promql/PromqlLogicalPlanOptimizerTests.java

@@ -101,7 +101,7 @@ public void testAvgAvgOverTimeOutput() {
         // | STATS AVG(AVG_OVER_TIME(`metrics.system.memory.utilization`)) BY host.name, TBUCKET(1h) | LIMIT 10000"
         var plan = planPromql("""
             TS k8s
-            | promql step 5m ( avg by (pod) (avg_over_time(network.bytes_in{pod=~"host-0|host-1|host-2"}[1h])) )
+            | promql step 1h ( avg by (pod) (avg_over_time(network.bytes_in{pod=~"host-0|host-1|host-2"}[1h])) )
             | LIMIT 1000
             """);
 
@@ -137,7 +137,7 @@ public void testPromqlAvgWithoutByDimension() {
         // | STATS AVG(AVG_OVER_TIME(`metrics.system.memory.utilization`)) BY TBUCKET(1h) | LIMIT 10000"
         var plan = planPromql("""
             TS k8s
-            | promql step 5m (
+            | promql step 1h (
                 avg(avg_over_time(network.bytes_in[1h]))
               )
             | LIMIT 1000
@@ -152,7 +152,7 @@ public void testRangeSelector() {
         // | STATS AVG(AVG_OVER_TIME(`metrics.system.memory.utilization`)) BY host.name, TBUCKET(1h) | LIMIT 10000"
         var plan = planPromql("""
             TS k8s
-            | promql step 10 ( max by (pod) (avg_over_time(network.bytes_in[1h])) )
+            | promql step 1h ( max by (pod) (avg_over_time(network.bytes_in[1h])) )
             """);
 
         System.out.println(plan);
@@ -164,7 +164,7 @@ public void testRate() {
         // | STATS AVG(RATE(`metrics.system.cpu.time`)) BY host.name, TBUCKET(1h) | LIMIT 10000"
         String testQuery = """
             TS k8s
-            | promql step 42 (
+            | promql step 1h (
                 avg by (pod) (rate(network.bytes_in[1h]))
                 )
             """;