[ES|QL] Rewrite RoundTo to QueryAndTags and merge range queries on RoundTo field

docs/changelog/132512.yaml

@@ -0,0 +1,5 @@
+pr: 132512
+summary: Rewrite `RoundTo` to `QueryAndTags`
+area: ES|QL
+type: enhancement
+issues: []

docs/changelog/132781.yaml

@@ -0,0 +1,6 @@
+pr: 132781
+summary: Rewrite `RoundTo` to `QueryAndTags` and merge range queries on `RoundTo`
+  field
+area: ES|QL
+type: enhancement
+issues: []

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/LocalPhysicalPlanOptimizer.java

@@ -18,6 +18,7 @@
 import org.elasticsearch.xpack.esql.optimizer.rules.physical.local.PushSampleToSource;
 import org.elasticsearch.xpack.esql.optimizer.rules.physical.local.PushStatsToSource;
 import org.elasticsearch.xpack.esql.optimizer.rules.physical.local.PushTopNToSource;
+import org.elasticsearch.xpack.esql.optimizer.rules.physical.local.ReplaceRoundToWithQueryAndTags;
 import org.elasticsearch.xpack.esql.optimizer.rules.physical.local.ReplaceSourceAttributes;
 import org.elasticsearch.xpack.esql.optimizer.rules.physical.local.SpatialDocValuesExtraction;
 import org.elasticsearch.xpack.esql.optimizer.rules.physical.local.SpatialShapeBoundsExtraction;
@@ -60,7 +61,7 @@ protected List<Batch<PhysicalPlan>> batches() {
     }
 
     protected static List<Batch<PhysicalPlan>> rules(boolean optimizeForEsSource) {
-        List<Rule<?, PhysicalPlan>> esSourceRules = new ArrayList<>(6);
+        List<Rule<?, PhysicalPlan>> esSourceRules = new ArrayList<>(7);
         esSourceRules.add(new ReplaceSourceAttributes());
         if (optimizeForEsSource) {
             esSourceRules.add(new PushTopNToSource());
@@ -74,6 +75,20 @@ protected static List<Batch<PhysicalPlan>> rules(boolean optimizeForEsSource) {
         // execute the rules multiple times to improve the chances of things being pushed down
         @SuppressWarnings("unchecked")
         var pushdown = new Batch<PhysicalPlan>("Push to ES", esSourceRules.toArray(Rule[]::new));
+        List<Rule<?, PhysicalPlan>> substitutionRules = new ArrayList<>(1);
+        if (optimizeForEsSource) {
+            substitutionRules.add(new ReplaceRoundToWithQueryAndTags());
+        }
+        // execute the SubstituteRoundToWithQueryAndTags rule once after all the other pushdown rules are applied, as this rule generate
+        // multiple QueryBuilders according the number of RoundTo points, it should be applied after all the other eligible pushdowns are
+        // done.
+        @SuppressWarnings("unchecked")
+        var substituteRoundToWithQueryAndTags = new Batch<PhysicalPlan>(
+            "Substitute RoundTo with QueryAndTags",
+            Limiter.ONCE,
+            substitutionRules.toArray(Rule[]::new)
+        );
+
         // add the field extraction in just one pass
         // add it at the end after all the other rules have ran
         var fieldExtraction = new Batch<>(
@@ -84,6 +99,6 @@ protected static List<Batch<PhysicalPlan>> rules(boolean optimizeForEsSource) {
             new SpatialShapeBoundsExtraction(),
             new ParallelizeTimeSeriesSource()
         );
-        return List.of(pushdown, fieldExtraction);
+        return List.of(pushdown, substituteRoundToWithQueryAndTags, fieldExtraction);
     }
 }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/physical/local/PushFiltersToSource.java

@@ -126,10 +126,10 @@ private static PhysicalPlan rewrite(
                 queryExec.indexMode(),
                 queryExec.indexNameWithModes(),
                 queryExec.output(),
-                query,
                 queryExec.limit(),
                 queryExec.sorts(),
-                queryExec.estimatedRowSize()
+                queryExec.estimatedRowSize(),
+                List.of(new EsQueryExec.QueryBuilderAndTags(query, List.of()))
             );
             // If the eval contains other aliases, not just field attributes, we need to keep them in the plan
             PhysicalPlan plan = evalFields.isEmpty() ? queryExec : new EvalExec(filterExec.source(), queryExec, evalFields);