ESQL: Opt into extra data stream resolution (#118378) (#118391)

* ESQL: Opt into extra data stream resolution

This opts ESQL's data node request into extra data stream resolution.

* Update docs/changelog/118378.yaml

docs/changelog/118378.yaml

@@ -0,0 +1,5 @@
+pr: 118378
+summary: Opt into extra data stream resolution
+area: ES|QL
+type: bug
+issues: []

x-pack/plugin/esql/qa/security/src/javaRestTest/java/org/elasticsearch/xpack/esql/EsqlSecurityIT.java

@@ -34,6 +34,7 @@ import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 
+import static org.elasticsearch.test.ListMatcher.matchesList;
 import static org.elasticsearch.test.MapMatcher.assertMap;
 import static org.elasticsearch.test.MapMatcher.matchesMap;
 import static org.hamcrest.Matchers.containsString;
@@ -56,6 +57,11 @@ public class EsqlSecurityIT extends ESRestTestCase {
         .user("metadata1_read2", "x-pack-test-password", "metadata1_read2", false)
         .user("alias_user1", "x-pack-test-password", "alias_user1", false)
         .user("alias_user2", "x-pack-test-password", "alias_user2", false)
+        .user("logs_foo_all", "x-pack-test-password", "logs_foo_all", false)
+        .user("logs_foo_16_only", "x-pack-test-password", "logs_foo_16_only", false)
+        .user("logs_foo_after_2021", "x-pack-test-password", "logs_foo_after_2021", false)
+        .user("logs_foo_after_2021_pattern", "x-pack-test-password", "logs_foo_after_2021_pattern", false)
+        .user("logs_foo_after_2021_alias", "x-pack-test-password", "logs_foo_after_2021_alias", false)
         .build();
 
     @Override
@@ -342,6 +348,14 @@ public class EsqlSecurityIT extends ESRestTestCase {
         assertThat(respMap.get("values"), equalTo(List.of(List.of(10.0))));
     }
 
+    public void testDocumentLevelSecurityFromStar() throws Exception {
+        Response resp = runESQLCommand("user3", "from in*x | stats sum=sum(value)");
+        assertOK(resp);
+        Map<String, Object> respMap = entityAsMap(resp);
+        assertThat(respMap.get("columns"), equalTo(List.of(Map.of("name", "sum", "type", "double"))));
+        assertThat(respMap.get("values"), equalTo(List.of(List.of(10.0))));
+    }
+
     public void testFieldLevelSecurityAllow() throws Exception {
         Response resp = runESQLCommand("fls_user", "FROM index* | SORT value | LIMIT 1");
         assertOK(resp);
@@ -545,6 +559,22 @@ public class EsqlSecurityIT extends ESRestTestCase {
         client().performRequest(new Request("DELETE", "_enrich/policy/songs"));
     }
 
+    public void testDataStream() throws IOException {
+        createDataStream();
+        MapMatcher twoResults = matchesMap().extraOk().entry("values", matchesList().item(matchesList().item(2)));
+        MapMatcher oneResult = matchesMap().extraOk().entry("values", matchesList().item(matchesList().item(1)));
+        assertMap(entityAsMap(runESQLCommand("logs_foo_all", "FROM logs-foo | STATS COUNT(*)")), twoResults);
+        assertMap(entityAsMap(runESQLCommand("logs_foo_16_only", "FROM logs-foo | STATS COUNT(*)")), oneResult);
+        assertMap(entityAsMap(runESQLCommand("logs_foo_after_2021", "FROM logs-foo | STATS COUNT(*)")), oneResult);
+        assertMap(entityAsMap(runESQLCommand("logs_foo_after_2021_pattern", "FROM logs-foo | STATS COUNT(*)")), oneResult);
+        assertMap(entityAsMap(runESQLCommand("logs_foo_after_2021_alias", "FROM alias-foo | STATS COUNT(*)")), oneResult);
+        assertMap(entityAsMap(runESQLCommand("logs_foo_all", "FROM logs-* | STATS COUNT(*)")), twoResults);
+        assertMap(entityAsMap(runESQLCommand("logs_foo_16_only", "FROM logs-* | STATS COUNT(*)")), oneResult);
+        assertMap(entityAsMap(runESQLCommand("logs_foo_after_2021", "FROM logs-* | STATS COUNT(*)")), oneResult);
+        assertMap(entityAsMap(runESQLCommand("logs_foo_after_2021_pattern", "FROM logs-* | STATS COUNT(*)")), oneResult);
+        assertMap(entityAsMap(runESQLCommand("logs_foo_after_2021_alias", "FROM alias-* | STATS COUNT(*)")), oneResult);
+    }
+
     protected Response runESQLCommand(String user, String command) throws IOException {
         if (command.toLowerCase(Locale.ROOT).contains("limit") == false) {
             // add a (high) limit to avoid warnings on default limit
@@ -592,4 +622,103 @@ public class EsqlSecurityIT extends ESRestTestCase {
         }
         return settings.build();
     }
+
+    private void createDataStream() throws IOException {
+        createDataStreamPolicy();
+        createDataStreamComponentTemplate();
+        createDataStreamIndexTemplate();
+        createDataStreamDocuments();
+        createDataStreamAlias();
+    }
+
+    private void createDataStreamPolicy() throws IOException {
+        Request request = new Request("PUT", "_ilm/policy/my-lifecycle-policy");
+        request.setJsonEntity("""
+            {
+              "policy": {
+                "phases": {
+                  "hot": {
+                    "actions": {
+                      "rollover": {
+                        "max_primary_shard_size": "50gb"
+                      }
+                    }
+                  },
+                  "delete": {
+                    "min_age": "735d",
+                    "actions": {
+                      "delete": {}
+                    }
+                  }
+                }
+              }
+            }""");
+        client().performRequest(request);
+    }
+
+    private void createDataStreamComponentTemplate() throws IOException {
+        Request request = new Request("PUT", "_component_template/my-template");
+        request.setJsonEntity("""
+            {
+                "template": {
+                   "settings": {
+                        "index.lifecycle.name": "my-lifecycle-policy"
+                   },
+                   "mappings": {
+                       "properties": {
+                           "@timestamp": {
+                               "type": "date",
+                               "format": "date_optional_time||epoch_millis"
+                           },
+                           "data_stream": {
+                               "properties": {
+                                   "namespace": {"type": "keyword"}
+                               }
+                           }
+                       }
+                   }
+                }
+            }""");
+        client().performRequest(request);
+    }
+
+    private void createDataStreamIndexTemplate() throws IOException {
+        Request request = new Request("PUT", "_index_template/my-index-template");
+        request.setJsonEntity("""
+            {
+                "index_patterns": ["logs-*"],
+                "data_stream": {},
+                "composed_of": ["my-template"],
+                "priority": 500
+            }""");
+        client().performRequest(request);
+    }
+
+    private void createDataStreamDocuments() throws IOException {
+        Request request = new Request("POST", "logs-foo/_bulk");
+        request.addParameter("refresh", "");
+        request.setJsonEntity("""
+            { "create" : {} }
+            { "@timestamp": "2099-05-06T16:21:15.000Z", "data_stream": {"namespace": "16"} }
+            { "create" : {} }
+            { "@timestamp": "2001-05-06T16:21:15.000Z", "data_stream": {"namespace": "17"} }
+            """);
+        assertMap(entityAsMap(client().performRequest(request)), matchesMap().extraOk().entry("errors", false));
+    }
+
+    private void createDataStreamAlias() throws IOException {
+        Request request = new Request("PUT", "_alias");
+        request.setJsonEntity("""
+            {
+              "actions": [
+                {
+                  "add": {
+                    "index": "logs-foo",
+                    "alias": "alias-foo"
+                  }
+                }
+              ]
+            }""");
+        assertMap(entityAsMap(client().performRequest(request)), matchesMap().extraOk().entry("errors", false));
+    }
 }

x-pack/plugin/esql/qa/security/src/javaRestTest/resources/roles.yml

@@ -92,3 +92,57 @@ fls_user:
       privileges: [ 'read' ]
       field_security:
         grant: [ value ]
+
+logs_foo_all:
+  cluster: []
+  indices:
+    - names: [ 'logs-foo' ]
+      privileges: [ 'read' ]
+
+logs_foo_16_only:
+  cluster: []
+  indices:
+    - names: [ 'logs-foo' ]
+      privileges: [ 'read' ]
+      query: |
+        {
+          "term": {
+            "data_stream.namespace": "16"
+          }
+        }
+
+logs_foo_after_2021:
+  cluster: []
+  indices:
+    - names: [ 'logs-foo' ]
+      privileges: [ 'read' ]
+      query: |
+        {
+          "range": {
+            "@timestamp": {"gte": "2021-01-01T00:00:00"}
+          }
+        }
+
+logs_foo_after_2021_pattern:
+  cluster: []
+  indices:
+    - names: [ 'logs-*' ]
+      privileges: [ 'read' ]
+      query: |
+        {
+          "range": {
+            "@timestamp": {"gte": "2021-01-01T00:00:00"}
+          }
+        }
+
+logs_foo_after_2021_alias:
+  cluster: []
+  indices:
+    - names: [ 'alias-foo' ]
+      privileges: [ 'read' ]
+      query: |
+        {
+          "range": {
+            "@timestamp": {"gte": "2021-01-01T00:00:00"}
+          }
+        }

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

@@ -118,6 +118,11 @@ final class DataNodeRequest extends TransportRequest implements IndicesRequest.R
         return this;
     }
 
+    @Override
+    public boolean includeDataStreams() {
+        return true;
+    }
+
     @Override
     public IndicesOptions indicesOptions() {
         return indicesOptions;