Add dimension values for time-series (#136052)

We optimized the VALUES aggregation for time-series queries rather than 
introducing a specialized version for dimension fields, and the VALUES
execution disappeared from the profiler. However, the VALUES aggregation
reappeared during recent testing with multi-valued dimension fields.
Therefore, this change introduces dimension_values for this purpose.

Author: dnhatn

server/src/main/resources/transport/definitions/referable/dimension_values.csv

@@ -0,0 +1 @@
+9197000,9185002

server/src/main/resources/transport/upper_bounds/9.2.csv

@@ -1 +1 @@
-esql_resolve_fields_response_created,9185001
+dimension_values,9185002

server/src/main/resources/transport/upper_bounds/9.3.csv

@@ -1 +1 @@
-ilm_downsample_force_merge,9196000
+dimension_values,9197000

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/aggregate/AggregateWritables.java

@@ -49,7 +49,8 @@ public static List<NamedWriteableRegistry.Entry> getNamedWriteables() {
             Present.ENTRY,
             PresentOverTime.ENTRY,
             Absent.ENTRY,
-            AbsentOverTime.ENTRY
+            AbsentOverTime.ENTRY,
+            DimensionValues.ENTRY
         );
     }
 }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/aggregate/DimensionValues.java

@@ -0,0 +1,84 @@
+/*
+ * 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.expression.function.aggregate;
+
+import org.elasticsearch.TransportVersion;
+import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.compute.aggregation.AggregatorFunctionSupplier;
+import org.elasticsearch.xpack.esql.core.expression.Expression;
+import org.elasticsearch.xpack.esql.core.expression.Literal;
+import org.elasticsearch.xpack.esql.core.tree.NodeInfo;
+import org.elasticsearch.xpack.esql.core.tree.Source;
+import org.elasticsearch.xpack.esql.core.type.DataType;
+import org.elasticsearch.xpack.esql.planner.ToAggregator;
+
+import java.io.IOException;
+import java.util.List;
+
+import static java.util.Collections.emptyList;
+
+/**
+ * A specialization of {@link Values} for collecting dimension fields in time-series queries.
+ */
+public class DimensionValues extends AggregateFunction implements ToAggregator {
+    public static final NamedWriteableRegistry.Entry ENTRY = new NamedWriteableRegistry.Entry(
+        Expression.class,
+        "DimensionValues",
+        DimensionValues::new
+    );
+
+    public static final TransportVersion DIMENSION_VALUES_VERSION = TransportVersion.fromName("dimension_values");
+
+    public DimensionValues(Source source, Expression field) {
+        super(source, field, Literal.TRUE, emptyList());
+    }
+
+    private DimensionValues(StreamInput in) throws IOException {
+        super(in);
+    }
+
+    @Override
+    public String getWriteableName() {
+        return ENTRY.name;
+    }
+
+    @Override
+    protected NodeInfo<DimensionValues> info() {
+        return NodeInfo.create(this, DimensionValues::new, field());
+    }
+
+    @Override
+    public DimensionValues replaceChildren(List<Expression> newChildren) {
+        return new DimensionValues(source(), newChildren.get(0));
+    }
+
+    @Override
+    public DimensionValues withFilter(Expression filter) {
+        if (filter instanceof Literal l && l.value() == Boolean.TRUE) {
+            return this;
+        }
+        throw new UnsupportedOperationException("Dimension values do not support filters");
+    }
+
+    @Override
+    public DataType dataType() {
+        return field().dataType().noText();
+    }
+
+    @Override
+    protected TypeResolution resolveType() {
+        return new Values(source(), field(), filter()).resolveType();
+    }
+
+    @Override
+    public AggregatorFunctionSupplier supplier() {
+        // TODO: link new implementation
+        return new Values(source(), field(), filter()).supplier();
+    }
+}

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/TranslateTimeSeriesAggregate.java

@@ -20,13 +20,15 @@
 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.DimensionValues;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.LastOverTime;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.TimeSeriesAggregateFunction;
 import org.elasticsearch.xpack.esql.expression.function.aggregate.Values;
 import org.elasticsearch.xpack.esql.expression.function.grouping.Bucket;
 import org.elasticsearch.xpack.esql.expression.function.grouping.TBucket;
 import org.elasticsearch.xpack.esql.expression.function.scalar.internal.PackDimension;
 import org.elasticsearch.xpack.esql.expression.function.scalar.internal.UnpackDimension;
+import org.elasticsearch.xpack.esql.optimizer.LogicalOptimizerContext;
 import org.elasticsearch.xpack.esql.plan.logical.Aggregate;
 import org.elasticsearch.xpack.esql.plan.logical.EsRelation;
 import org.elasticsearch.xpack.esql.plan.logical.Eval;
@@ -63,16 +65,16 @@
  * becomes
  *
  * TS k8s
- * | STATS rate_$1=rate(request), VALUES(host) BY _tsid
- * | STATS max(rate_$1) BY host=`VALUES(host)`
+ * | STATS rate_$1=rate(request), DIMENSION_VALUES(host) BY _tsid
+ * | STATS max(rate_$1) BY host=`DIMENSION_VALUES(host)`
  *
  * TS k8s | STATS avg(rate(request)) BY host
  *
  * becomes
  *
  * TS k8s
- * | STATS rate_$1=rate(request), VALUES(host) BY _tsid
- * | STATS sum(rate_$1), count(rate_$1) BY host=`VALUES(host)`
+ * | STATS rate_$1=rate(request), DIMENSION_VALUES(host) BY _tsid
+ * | STATS sum(rate_$1), count(rate_$1) BY host=`DIMENSION_VALUES(host)`
  * | EVAL `avg(rate(request))` = `sum(rate_$1)` / `count(rate_$1)`
  * | KEEP `avg(rate(request))`, host
  *
@@ -82,8 +84,8 @@
  *
  * TS k8s
  * | EVAL  `bucket(@timestamp, 1minute)`=datetrunc(@timestamp, 1minute)
- * | STATS rate_$1=rate(request), VALUES(host) BY _tsid,`bucket(@timestamp, 1minute)`
- * | STATS sum=sum(rate_$1), count(rate_$1) BY host=`VALUES(host)`, `bucket(@timestamp, 1minute)`
+ * | STATS rate_$1=rate(request), DIMENSION_VALUES(host) BY _tsid,`bucket(@timestamp, 1minute)`
+ * | STATS sum=sum(rate_$1), count(rate_$1) BY host=`DIMENSION_VALUES(host)`, `bucket(@timestamp, 1minute)`
  * | EVAL `avg(rate(request))` = `sum(rate_$1)` / `count(rate_$1)`
  * | KEEP `avg(rate(request))`, host, `bucket(@timestamp, 1minute)`
  * </pre>
@@ -146,22 +148,16 @@
  * | STATS max(rate_$1 + rate_$2) BY host_values, time_bucket
  * </pre>
  */
-public final class TranslateTimeSeriesAggregate extends OptimizerRules.OptimizerRule<Aggregate> {
+public final class TranslateTimeSeriesAggregate extends OptimizerRules.ParameterizedOptimizerRule<
+    TimeSeriesAggregate,
+    LogicalOptimizerContext> {
 
     public TranslateTimeSeriesAggregate() {
         super(OptimizerRules.TransformDirection.UP);
     }
 
     @Override
-    protected LogicalPlan rule(Aggregate aggregate) {
-        if (aggregate instanceof TimeSeriesAggregate ts && ts.timeBucket() == null) {
-            return translate(ts);
-        } else {
-            return aggregate;
-        }
-    }
-
-    LogicalPlan translate(TimeSeriesAggregate aggregate) {
+    protected LogicalPlan rule(TimeSeriesAggregate aggregate, LogicalOptimizerContext context) {
         Holder<Attribute> tsid = new Holder<>();
         Holder<Attribute> timestamp = new Holder<>();
         aggregate.forEachDown(EsRelation.class, r -> {
@@ -278,7 +274,7 @@ LogicalPlan translate(TimeSeriesAggregate aggregate) {
                 firstPassGroupings.add(newFinalGroup);
                 secondPassGroupings.add(new Alias(g.source(), g.name(), newFinalGroup.toAttribute(), g.id()));
             } else {
-                var valuesAgg = new Alias(g.source(), g.name(), new Values(g.source(), g));
+                var valuesAgg = new Alias(g.source(), g.name(), valuesAggregate(context, g));
                 firstPassAggs.add(valuesAgg);
                 Alias pack = new Alias(
                     g.source(),
@@ -361,6 +357,14 @@ private static List<? extends NamedExpression> mergeExpressions(
         return merged;
     }
 
+    private AggregateFunction valuesAggregate(LogicalOptimizerContext context, Attribute group) {
+        if (group.isDimension() && context.minimumVersion().supports(DimensionValues.DIMENSION_VALUES_VERSION)) {
+            return new DimensionValues(group.source(), group);
+        } else {
+            return new Values(group.source(), group);
+        }
+    }
+
     private static class InternalNames {
         final Map<String, Integer> next = new HashMap<>();
 

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/aggregate/DimensionValuesSerializationTests.java

@@ -0,0 +1,27 @@
+/*
+ * 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.expression.function.aggregate;
+
+import org.elasticsearch.xpack.esql.expression.AbstractExpressionSerializationTests;
+
+import java.io.IOException;
+
+public class DimensionValuesSerializationTests extends AbstractExpressionSerializationTests<DimensionValues> {
+    @Override
+    protected DimensionValues createTestInstance() {
+        return new DimensionValues(randomSource(), randomChild());
+    }
+
+    @Override
+    protected DimensionValues mutateInstance(DimensionValues instance) throws IOException {
+        return new DimensionValues(
+            instance.source(),
+            randomValueOtherThan(instance.field(), AbstractExpressionSerializationTests::randomChild)
+        );
+    }
+}

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

@@ -7,6 +7,7 @@
 
 package org.elasticsearch.xpack.esql.optimizer;
 
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.index.IndexMode;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xpack.esql.EsqlTestUtils;
@@ -41,6 +42,8 @@ public abstract class AbstractLogicalPlanOptimizerTests extends ESTestCase {
     protected static LogicalOptimizerContext logicalOptimizerCtx;
     protected static LogicalPlanOptimizer logicalOptimizer;
 
+    protected static LogicalPlanOptimizer logicalOptimizerWithLatestVersion;
+
     protected static Map<String, EsField> mapping;
     protected static Analyzer analyzer;
     protected static Map<String, EsField> mappingAirports;
@@ -74,6 +77,9 @@ public static void init() {
         parser = new EsqlParser();
         logicalOptimizerCtx = unboundLogicalOptimizerContext();
         logicalOptimizer = new LogicalPlanOptimizer(logicalOptimizerCtx);
+        logicalOptimizerWithLatestVersion = new LogicalPlanOptimizer(
+            new LogicalOptimizerContext(logicalOptimizerCtx.configuration(), logicalOptimizerCtx.foldCtx(), TransportVersion.current())
+        );
         enrichResolution = new EnrichResolution();
         AnalyzerTestUtils.loadEnrichPolicyResolution(enrichResolution, "languages_idx", "id", "languages_idx", "mapping-languages.json");