Fix NPE in TimeSeriesExtractFieldOperator (#131497)

We miss checking whether a field exists when populating the dimension 
attributes. This issue occurs when a field exists in some, but not all
target indices.

Author: dnhatn

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/lucene/read/TimeSeriesExtractFieldOperator.java

@@ -254,7 +254,8 @@ static final class ShardLevelFieldsReader implements Releasable {
             this.storedFieldsSpec = storedFieldsSpec;
             this.dimensions = new boolean[fields.size()];
             for (int i = 0; i < fields.size(); i++) {
-                dimensions[i] = shardContext.fieldType(fields.get(i).name()).isDimension();
+                final var mappedFieldType = shardContext.fieldType(fields.get(i).name());
+                dimensions[i] = mappedFieldType != null && mappedFieldType.isDimension();
             }
         }
 

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/TimeSeriesIT.java

@@ -10,6 +10,7 @@
 import org.elasticsearch.Build;
 import org.elasticsearch.common.Randomness;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.compute.lucene.TimeSeriesSourceOperator;
 import org.elasticsearch.compute.operator.DriverProfile;
 import org.elasticsearch.compute.operator.TimeSeriesAggregationOperator;
@@ -56,7 +57,7 @@ public void testEmpty() {
         run("TS empty_index | LIMIT 1").close();
     }
 
-    record Doc(String host, String cluster, long timestamp, int requestCount, double cpu) {}
+    record Doc(String host, String cluster, long timestamp, int requestCount, double cpu, ByteSizeValue memory) {}
 
     final List<Doc> docs = new ArrayList<>();
 
@@ -84,7 +85,6 @@ static Double computeRate(List<RequestCounter> values) {
 
     @Before
     public void populateIndex() {
-        // this can be expensive, do one
         Settings settings = Settings.builder().put("mode", "time_series").putList("routing_path", List.of("host", "cluster")).build();
         client().admin()
             .indices()
@@ -99,6 +99,8 @@ public void populateIndex() {
                 "type=keyword,time_series_dimension=true",
                 "cpu",
                 "type=double,time_series_metric=gauge",
+                "memory",
+                "type=long,time_series_metric=gauge",
                 "request_count",
                 "type=integer,time_series_metric=counter"
             )
@@ -123,7 +125,8 @@ public void populateIndex() {
                     }
                 });
                 int cpu = randomIntBetween(0, 100);
-                docs.add(new Doc(host, hostToClusters.get(host), timestamp, requestCount, cpu));
+                ByteSizeValue memory = ByteSizeValue.ofBytes(randomIntBetween(1024, 1024 * 1024));
+                docs.add(new Doc(host, hostToClusters.get(host), timestamp, requestCount, cpu, memory));
             }
         }
         Randomness.shuffle(docs);
@@ -138,6 +141,8 @@ public void populateIndex() {
                     doc.cluster,
                     "cpu",
                     doc.cpu,
+                    "memory",
+                    doc.memory.getBytes(),
                     "request_count",
                     doc.requestCount
                 )
@@ -417,6 +422,63 @@ public void testIndexMode() {
         assertThat(failure.getMessage(), containsString("Unknown index [hosts-old]"));
     }
 
+    public void testFieldDoesNotExist() {
+        // the old-hosts index doesn't have the cpu field
+        Settings settings = Settings.builder().put("mode", "time_series").putList("routing_path", List.of("host", "cluster")).build();
+        client().admin()
+            .indices()
+            .prepareCreate("old-hosts")
+            .setSettings(settings)
+            .setMapping(
+                "@timestamp",
+                "type=date",
+                "host",
+                "type=keyword,time_series_dimension=true",
+                "cluster",
+                "type=keyword,time_series_dimension=true",
+                "memory",
+                "type=long,time_series_metric=gauge",
+                "request_count",
+                "type=integer,time_series_metric=counter"
+            )
+            .get();
+        Randomness.shuffle(docs);
+        for (Doc doc : docs) {
+            client().prepareIndex("old-hosts")
+                .setSource(
+                    "@timestamp",
+                    doc.timestamp,
+                    "host",
+                    doc.host,
+                    "cluster",
+                    doc.cluster,
+                    "memory",
+                    doc.memory.getBytes(),
+                    "request_count",
+                    doc.requestCount
+                )
+                .get();
+        }
+        client().admin().indices().prepareRefresh("old-hosts").get();
+        try (var resp1 = run("""
+            TS hosts,old-hosts
+            | STATS sum(rate(request_count)), max(last_over_time(cpu)), max(last_over_time(memory)) BY cluster, host
+            | SORT cluster, host
+            | DROP `sum(rate(request_count))`
+            """)) {
+            try (var resp2 = run("""
+                TS hosts
+                | STATS sum(rate(request_count)), max(last_over_time(cpu)), max(last_over_time(memory)) BY cluster, host
+                | SORT cluster, host
+                | DROP `sum(rate(request_count))`
+                """)) {
+                List<List<Object>> values1 = EsqlTestUtils.getValuesList(resp1);
+                List<List<Object>> values2 = EsqlTestUtils.getValuesList(resp2);
+                assertThat(values1, equalTo(values2));
+            }
+        }
+    }
+
     public void testProfile() {
         EsqlQueryRequest request = new EsqlQueryRequest();
         request.profile(true);