ESQL: GROUP BY ALL

x-pack/plugin/esql/qa/testFixtures/src/main/resources/k8s-timeseries.csv-spec

@@ -721,3 +721,24 @@ mx:integer | tbucket:datetime
 ;
 
 
+
+
+Group by all for over time agg with no wrapper, no time bucket  
+required_capability: metrics_command  
+required_capability: metrics_group_by_all
+
+TS k8s   
+| STATS count = count_over_time(network.cost) 
+;  
+
+count:long | cluster:keyword | pod:keyword  
+13         | staging         | one
+20         | staging         | two
+24         | staging         | three
+29         | qa              | one
+29         | qa              | two
+24         | qa              | three
+23         | prod            | one
+16         | prod            | two
+22         | prod            | three
+;  

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java

@@ -1033,6 +1033,10 @@ public enum Cap {
          */
         @Deprecated
         METRICS_COMMAND(Build.current().isSnapshot()),
+        /**
+         * Enables automatically grouping by all dimension fields in TS mode queries
+         */
+        METRICS_GROUP_BY_ALL(),
 
         /**
          * Are the {@code documents_found} and {@code values_loaded} fields available

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/analysis/Analyzer.java

@@ -212,7 +212,8 @@ public class Analyzer extends ParameterizedRuleExecutor<LogicalPlan, AnalyzerCon
             new ResolveLookupTables(),
             new ResolveFunctions(),
             new ResolveInference(),
-            new DateMillisToNanosInEsRelation()
+            new DateMillisToNanosInEsRelation(),
+            new TimeSeriesGroupByAll()
         ),
         new Batch<>(
             "Resolution",
@@ -1235,7 +1236,7 @@ private LogicalPlan resolveKeep(Project p, List<Attribute> childOutput) {
         private LogicalPlan resolveDrop(Drop drop, List<Attribute> childOutput) {
             List<NamedExpression> resolvedProjections = new ArrayList<>(childOutput);
 
-            for (var ne : drop.removals()) {
+            for (NamedExpression ne : drop.removals()) {
                 List<? extends NamedExpression> resolved;
 
                 if (ne instanceof UnresolvedNamePattern np) {

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/analysis/TimeSeriesGroupByAll.java

@@ -0,0 +1,150 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.esql.analysis;
+
+import org.elasticsearch.index.IndexMode;
+import org.elasticsearch.xpack.esql.core.expression.Alias;
+import org.elasticsearch.xpack.esql.core.expression.Attribute;
+import org.elasticsearch.xpack.esql.core.expression.Expression;
+import org.elasticsearch.xpack.esql.core.expression.MetadataAttribute;
+import org.elasticsearch.xpack.esql.core.expression.NamedExpression;
+import org.elasticsearch.xpack.esql.core.type.DataType;
+import org.elasticsearch.xpack.esql.core.util.CollectionUtils;
+import org.elasticsearch.xpack.esql.core.util.Holder;
+import org.elasticsearch.xpack.esql.expression.function.aggregate.AggregateFunction;
+import org.elasticsearch.xpack.esql.expression.function.aggregate.Rate;
+import org.elasticsearch.xpack.esql.expression.function.aggregate.TimeSeriesAggregateFunction;
+import org.elasticsearch.xpack.esql.expression.function.aggregate.Values;
+import org.elasticsearch.xpack.esql.plan.logical.EsRelation;
+import org.elasticsearch.xpack.esql.plan.logical.LogicalPlan;
+import org.elasticsearch.xpack.esql.plan.logical.TimeSeriesAggregate;
+import org.elasticsearch.xpack.esql.rule.Rule;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * This rule implements the "group by all" logic for time series aggregations.  It is intended to work in conjunction with
+ * {@link org.elasticsearch.xpack.esql.optimizer.rules.logical.TranslateTimeSeriesAggregate}, and should be run before that
+ * rule.  This rule adds output columns corresponding to the dimensions on the indices involved in the query, as discovered
+ * by the {@link org.elasticsearch.xpack.esql.session.IndexResolver}. Despite the name, this does not acutally group on the
+ * dimension values, for efficiency reasons.
+ * <p>
+ * This rule will operate on "bare" over time aggregations,
+ */
+public class TimeSeriesGroupByAll extends Rule<LogicalPlan, LogicalPlan> {
+    @Override
+    public LogicalPlan apply(LogicalPlan logicalPlan) {
+        return logicalPlan.transformUp(node -> node instanceof TimeSeriesAggregate, this::rule);
+    }
+
+    public LogicalPlan rule(TimeSeriesAggregate aggregate) {
+        // Flag to check if we should apply this rule.
+        boolean hasTopLevelOverTimeAggs = false;
+        Holder<Boolean> hasRateAggregates = new Holder<>(Boolean.FALSE);
+        // the new `Value(dimension)` aggregation functions we intend to add to the query, along with the translated over time aggs
+        List<NamedExpression> newAggregateFunctions = new ArrayList<>();
+        for (NamedExpression agg : aggregate.aggregates()) {
+            // We assume that all the aggregate functions in aggregates are wrapped as Aliases. I don't know why
+            // this should be the case, but it is ubiquitous throughout the planner.
+            if (agg instanceof Alias alias && alias.child() instanceof AggregateFunction af) {
+                if (af instanceof TimeSeriesAggregateFunction tsAgg) {
+                    hasTopLevelOverTimeAggs = true;
+                    newAggregateFunctions.add(new Alias(alias.source(), alias.name(), tsAgg.perTimeSeriesAggregation()));
+                }
+                af.forEachDown(TimeSeriesAggregateFunction.class, tsAgg -> {
+                    if (tsAgg instanceof Rate) {
+                        hasRateAggregates.set(Boolean.TRUE);
+                    }
+                });
+                // TODO: Deal with mixed top level and wrapped aggs case
+            }
+        }
+        if (hasTopLevelOverTimeAggs == false) {
+            // If there are no top level time series aggregations, there's no need for this rule to apply
+            return aggregate;
+        }
+
+        // Grouping parameters for the new aggregation node.
+        List<Expression> groupings = new ArrayList<>();
+
+        Holder<Attribute> tsid = new Holder<>();
+        Holder<Attribute> timestamp = new Holder<>();
+        Set<Attribute> dimensions = new HashSet<>();
+        getTsFields(aggregate, tsid, timestamp, dimensions);
+        if (tsid.get() == null) {
+            tsid.set(new MetadataAttribute(aggregate.source(), MetadataAttribute.TSID_FIELD, DataType.KEYWORD, false));
+        }
+        if (timestamp.get() == null) {
+            throw new IllegalArgumentException("_tsid or @timestamp field are missing from the time-series source");
+        }
+
+        // NOCOMMIT - this is redundant with behavior in TranslateTSA
+        // NOCOMMIT - This behavior is tangential to this rule; maybe we should break it out into a separate rule?
+        // Add the _tsid to the EsRelation Leaf, if it's not there already
+        LogicalPlan newChild = aggregate.child().transformUp(EsRelation.class, r -> {
+            IndexMode indexMode = hasRateAggregates.get() ? r.indexMode() : IndexMode.STANDARD;
+            if (r.output().contains(tsid.get()) == false) {
+                return new EsRelation(
+                    r.source(),
+                    r.indexPattern(),
+                    indexMode,
+                    r.indexNameWithModes(),
+                    CollectionUtils.combine(r.output(), tsid.get())
+                );
+            } else {
+                return new EsRelation(r.source(), r.indexPattern(), indexMode, r.indexNameWithModes(), r.output());
+            }
+        });
+        // Group the new aggregations by tsid. This is equivalent to grouping by all dimensions.
+        // NOTE - we do not add the tsid to the aggregates list because we do not want to include it in the final output
+        groupings.add(tsid.get());
+        // Add the time bucket grouping for the new agg
+        // NOCOMMIT - validation rule for groupings?
+        groupings.addAll(aggregate.groupings());
+        for (Attribute dimension : dimensions) {
+            // We add the dimensions as Values aggs here as an optimization. Grouping by the _tsid should already ensure
+            // one row per unique combination of dimensions, and collecting those values in a Values aggregation is less
+            // computation than hashing them for a grouping operation.
+            // NOCOMMIT - This is also redundant. TranslateTSA already turns all the inner groupings into values aggs. Just do this there.
+            newAggregateFunctions.add(new Alias(dimension.source(), dimension.name(), new Values(dimension.source(), dimension)));
+        }
+        TimeSeriesAggregate newAggregate = new TimeSeriesAggregate(
+            aggregate.source(),
+            newChild,
+            groupings,
+            newAggregateFunctions,
+            null,
+            true
+        );
+        return newAggregate;
+    }
+
+    private static void getTsFields(
+        TimeSeriesAggregate aggregate,
+        Holder<Attribute> tsid,
+        Holder<Attribute> timestamp,
+        Set<Attribute> dimensions
+    ) {
+        aggregate.forEachDown(EsRelation.class, r -> {
+            for (Attribute attr : r.output()) {
+                if (attr.name().equals(MetadataAttribute.TSID_FIELD)) {
+                    tsid.set(attr);
+                }
+                if (attr.name().equals(MetadataAttribute.TIMESTAMP_FIELD)) {
+                    timestamp.set(attr);
+                }
+                if (attr.isDimension()) {
+                    dimensions.add(attr);
+                }
+            }
+        });
+    }
+}

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

@@ -17,6 +17,12 @@
 import java.util.List;
 import java.util.Objects;
 
+/**
+ * Drop is an intermediary object used during the {@link org.elasticsearch.xpack.esql.analysis.Analyzer} phase of query planning.
+ * DROP commands are parsed into Drop objects, which the {@link org.elasticsearch.xpack.esql.analysis.Analyzer.ResolveRefs} rule then
+ * rewrites into {@link org.elasticsearch.xpack.esql.plan.logical.local.EsqlProject} plans, along with other projection-like commands.
+ * As such, Drop is neither serializable nor able to be mapped to a corresponding physical plan.
+ */
 public class Drop extends UnaryPlan implements TelemetryAware, Streaming, SortAgnostic {
     private final List<NamedExpression> removals;
 

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

@@ -45,20 +45,36 @@ public class TimeSeriesAggregate extends Aggregate {
 
     private final Bucket timeBucket;
 
+    private final boolean hasTopLevelOverTimeFunctions;
+
     public TimeSeriesAggregate(
         Source source,
         LogicalPlan child,
         List<Expression> groupings,
         List<? extends NamedExpression> aggregates,
         Bucket timeBucket
+    ) {
+        this(source, child, groupings, aggregates, timeBucket, false);
+    }
+
+    public TimeSeriesAggregate(
+        Source source,
+        LogicalPlan child,
+        List<Expression> groupings,
+        List<? extends NamedExpression> aggregates,
+        Bucket timeBucket,
+        boolean hasTopLevelOverTimeFunctions
     ) {
         super(source, child, groupings, aggregates);
         this.timeBucket = timeBucket;
+        this.hasTopLevelOverTimeFunctions = hasTopLevelOverTimeFunctions;
     }
 
     public TimeSeriesAggregate(StreamInput in) throws IOException {
         super(in);
         this.timeBucket = in.readOptionalWriteable(inp -> (Bucket) Bucket.ENTRY.reader.read(inp));
+        // Shouldn't need to be serialized; at least not yet
+        this.hasTopLevelOverTimeFunctions = false;
     }
 
     @Override
@@ -67,24 +83,28 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeOptionalWriteable(timeBucket);
     }
 
+    public boolean hasTopLevelOverTimeFunctions() {
+        return hasTopLevelOverTimeFunctions;
+    }
+
     @Override
     public String getWriteableName() {
         return ENTRY.name;
     }
 
     @Override
     protected NodeInfo<Aggregate> info() {
-        return NodeInfo.create(this, TimeSeriesAggregate::new, child(), groupings, aggregates, timeBucket);
+        return NodeInfo.create(this, TimeSeriesAggregate::new, child(), groupings, aggregates, timeBucket, hasTopLevelOverTimeFunctions);
     }
 
     @Override
     public TimeSeriesAggregate replaceChild(LogicalPlan newChild) {
-        return new TimeSeriesAggregate(source(), newChild, groupings, aggregates, timeBucket);
+        return new TimeSeriesAggregate(source(), newChild, groupings, aggregates, timeBucket, hasTopLevelOverTimeFunctions);
     }
 
     @Override
     public TimeSeriesAggregate with(LogicalPlan child, List<Expression> newGroupings, List<? extends NamedExpression> newAggregates) {
-        return new TimeSeriesAggregate(source(), child, newGroupings, newAggregates, timeBucket);
+        return new TimeSeriesAggregate(source(), child, newGroupings, newAggregates, timeBucket, hasTopLevelOverTimeFunctions);
     }
 
     @Override

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

@@ -0,0 +1,48 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.esql.optimizer;
+
+import org.elasticsearch.xpack.esql.action.EsqlCapabilities;
+import org.elasticsearch.xpack.esql.expression.function.aggregate.Count;
+import org.elasticsearch.xpack.esql.expression.function.aggregate.Values;
+import org.elasticsearch.xpack.esql.plan.physical.PhysicalPlan;
+import org.elasticsearch.xpack.esql.plan.physical.TimeSeriesAggregateExec;
+import org.elasticsearch.xpack.esql.session.Configuration;
+
+import java.util.List;
+
+import static org.elasticsearch.xpack.esql.EsqlTestUtils.as;
+
+public class TimeseriesGroupByAllPhysicalPlannerTests extends LocalPhysicalPlanOptimizerTests {
+    public TimeseriesGroupByAllPhysicalPlannerTests(String name, Configuration config) {
+        super(name, config);
+    }
+
+    public void testSimple() {
+        assumeTrue("requires metrics command", EsqlCapabilities.Cap.METRICS_COMMAND.isEnabled());
+        assumeTrue("requires metrics group by all", EsqlCapabilities.Cap.METRICS_GROUP_BY_ALL.isEnabled());
+
+        String testQuery = """
+            TS k8s
+            | STATS count = count_over_time(network.cost)
+            | LIMIT 10
+            """;
+        PhysicalPlan actualPlan = plannerOptimizerTimeSeries.plan(testQuery);
+
+        List<PhysicalPlan> aggs = actualPlan.collect(node -> node instanceof TimeSeriesAggregateExec);
+        for (PhysicalPlan agg : aggs) {
+            if (agg instanceof TimeSeriesAggregateExec tsAgg) {
+                assertEquals(3, tsAgg.aggregates().size());
+                // Check the first child to unwrap the alias
+                Count countFn = as(tsAgg.aggregates().get(0).children().get(0), Count.class);
+                Values clusterFn = as(tsAgg.aggregates().get(1).children().get(0), Values.class);
+                Values podFn = as(tsAgg.aggregates().get(2).children().get(0), Values.class);
+            }
+        }
+    }
+}