Restrict sorting for _source and counter field types

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/type/DataType.java

@@ -425,6 +425,10 @@ public static boolean isRepresentable(DataType t) {
             && t.isCounter() == false;
     }
 
+    public static boolean isCounter(DataType t) {
+        return t == COUNTER_DOUBLE || t == COUNTER_INTEGER || t == COUNTER_LONG;
+    }
+
     public static boolean isSpatialPoint(DataType t) {
         return t == GEO_POINT || t == CARTESIAN_POINT;
     }
@@ -437,6 +441,10 @@ public static boolean isSpatial(DataType t) {
         return t == GEO_POINT || t == CARTESIAN_POINT || t == GEO_SHAPE || t == CARTESIAN_SHAPE;
     }
 
+    public static boolean isSortable(DataType t) {
+        return false == (t == SOURCE || isCounter(t) || isSpatial(t));
+    }
+
     public String nameUpper() {
         return name;
     }

x-pack/plugin/esql/qa/testFixtures/src/main/resources/tsdb-mapping.json

@@ -27,6 +27,10 @@
         "message_in": {
           "type": "float",
           "time_series_metric": "counter"
+        },
+        "message_out": {
+          "type": "integer",
+          "time_series_metric": "counter"
         }
       }
     }

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

@@ -183,7 +183,7 @@ else if (p instanceof Lookup lookup) {
 
             checkOperationsOnUnsignedLong(p, failures);
             checkBinaryComparison(p, failures);
-            checkForSortOnSpatialTypes(p, failures);
+            checkForSortableDataTypes(p, failures);
 
             checkFilterMatchConditions(p, failures);
             checkFullTextQueryFunctions(p, failures);
@@ -548,12 +548,12 @@ private static Failure validateUnsignedLongNegation(Neg neg) {
     }
 
     /**
-     * Makes sure that spatial types do not appear in sorting contexts.
+     * Some datatypes care not sortable
      */
-    private static void checkForSortOnSpatialTypes(LogicalPlan p, Set<Failure> localFailures) {
+    private static void checkForSortableDataTypes(LogicalPlan p, Set<Failure> localFailures) {
         if (p instanceof OrderBy ob) {
             ob.order().forEach(order -> {
-                if (DataType.isSpatial(order.dataType())) {
+                if (DataType.isSortable(order.dataType()) == false) {
                     localFailures.add(fail(order, "cannot sort on " + order.dataType().typeName()));
                 }
             });

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

@@ -115,7 +115,7 @@ public DataType dataType() {
     protected TypeResolution resolveType() {
         TypeResolution resolution = isType(
             field(),
-            dt -> dt == DataType.COUNTER_LONG || dt == DataType.COUNTER_INTEGER || dt == DataType.COUNTER_DOUBLE,
+            dt -> DataType.isCounter(dt),
             sourceText(),
             FIRST,
             "counter_long",

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

@@ -25,12 +25,16 @@
 import java.util.LinkedHashMap;
 import java.util.LinkedHashSet;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
 
 import static org.elasticsearch.xpack.esql.EsqlTestUtils.paramAsConstant;
 import static org.elasticsearch.xpack.esql.EsqlTestUtils.withDefaultLimitWarning;
 import static org.elasticsearch.xpack.esql.analysis.AnalyzerTestUtils.loadMapping;
+import static org.elasticsearch.xpack.esql.core.type.DataType.COUNTER_DOUBLE;
+import static org.elasticsearch.xpack.esql.core.type.DataType.COUNTER_INTEGER;
+import static org.elasticsearch.xpack.esql.core.type.DataType.COUNTER_LONG;
 import static org.elasticsearch.xpack.esql.core.type.DataType.UNSIGNED_LONG;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
@@ -908,6 +912,25 @@ public void testSpatialSort() {
         assertEquals("1:42: cannot sort on cartesian_shape", error("FROM countries_bbox_web | LIMIT 5 | sort shape", countriesBboxWeb));
     }
 
+    public void testSourceSorting() {
+        assertEquals("1:35: cannot sort on _source", error("from test metadata _source | sort _source"));
+    }
+
+    public void testCountersSorting() {
+        Map<DataType, String> counterDataTypes = Map.of(
+            COUNTER_DOUBLE,
+            "network.message_in",
+            COUNTER_INTEGER,
+            "network.message_out",
+            COUNTER_LONG,
+            "network.bytes_out"
+        );
+        for (DataType counterDT : counterDataTypes.keySet()) {
+            var fieldName = counterDataTypes.get(counterDT);
+            assertEquals("1:18: cannot sort on " + counterDT.name().toLowerCase(Locale.ROOT), error("from test | sort " + fieldName, tsdb));
+        }
+    }
+
     public void testInlineImpossibleConvert() {
         assertEquals("1:5: argument of [false::ip] must be [ip or string], found value [false] type [boolean]", error("ROW false::ip"));
     }

x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/esql/140_metadata.yml

@@ -155,3 +155,11 @@ setup:
       esql.query:
         body:
           query: 'FROM test [metadata _source] | STATS COUNT_DISTINCT(_source)'
+
+---
+"sort on _source not allowed":
+  - do:
+      catch: /cannot sort on _source/
+      esql.query:
+        body:
+          query: 'FROM test metadata _source | sort _source'

x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/esql/40_tsdb.yml

@@ -179,10 +179,10 @@ cast counter then filter:
 ---
 sort on counter without cast:
   - do:
-      catch: bad_request
+      catch: /cannot sort on counter_long/
       esql.query:
         body:
-          query: 'from test |  KEEP k8s.pod.network.tx | sort @k8s.pod.network.tx | limit 1'
+          query: 'from test |  KEEP k8s.pod.network.tx | sort k8s.pod.network.tx | limit 1'
 
 ---
 cast then sort on counter: