Refactor to access views within FROM command

Also add csv-spec tests that actually pass!

Author: craigtaverner

x-pack/plugin/esql/qa/server/src/main/java/org/elasticsearch/xpack/esql/qa/rest/EsqlSpecTestCase.java

@@ -51,6 +51,7 @@
 import java.util.Locale;
 import java.util.Map;
 import java.util.TreeMap;
+import java.util.concurrent.Callable;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import java.util.stream.LongStream;
@@ -121,20 +122,49 @@ protected EsqlSpecTestCase(
         this.mode = randomFrom(Mode.values());
     }
 
-    private static boolean dataLoaded = false;
+    private static final Object lock = new Object();
+    private static volatile boolean dataLoaded = false;
+    private static volatile boolean setupStarted = false;
+    private static volatile Throwable setupFailure = null;
 
     @Before
-    public void setup() throws IOException {
-        boolean supportsLookup = supportsIndexModeLookup();
-        boolean supportsSourceMapping = supportsSourceFieldMapping();
-        boolean supportsInferenceTestService = supportsInferenceTestService();
-        if (dataLoaded == false) {
+    public void setup() {
+        protectedBlock(() -> {
+            boolean supportsLookup = supportsIndexModeLookup();
+            boolean supportsSourceMapping = supportsSourceFieldMapping();
+            boolean supportsInferenceTestService = supportsInferenceTestService();
             if (supportsInferenceTestService) {
                 createInferenceEndpoints(adminClient());
             }
-
             loadDataSetIntoEs(client(), supportsLookup, supportsSourceMapping, supportsInferenceTestService);
-            dataLoaded = true;
+            return null;
+        });
+    }
+
+    private static void protectedBlock(Callable<Void> callable) {
+        if (dataLoaded) {
+            return;
+        }
+        // In case tests get run in parallel, we ensure only one setup is run, and other tests wait for this
+        synchronized (lock) {
+            if (dataLoaded) {
+                return;
+            }
+            if (setupStarted) {
+                // Should only happen if a previous test setup failed, possibly with partial setup, let's fail fast the current test
+                if (setupFailure != null) {
+                    fail(setupFailure, "Previous test setup failed: " + setupFailure.getMessage());
+                }
+                fail("Previous test setup failed with unknown error");
+            }
+            setupStarted = true;
+            try {
+                callable.call();
+                dataLoaded = true;
+            } catch (Throwable t) {
+                setupFailure = t;
+                fail(setupFailure, "Current test setup failed: " + setupFailure.getMessage());
+            }
         }
     }
 

x-pack/plugin/esql/qa/testFixtures/src/main/java/org/elasticsearch/xpack/esql/CsvTestsDataLoader.java

@@ -268,6 +268,11 @@ public class CsvTestsDataLoader {
     );
     public static final String NUMERIC_REGEX = "-?\\d+(\\.\\d+)?";
 
+    private static final ViewConfig ADDRESS_COUNTRIES = new ViewConfig("address_countries");
+    private static final ViewConfig AIRPORTS_COUNTRIES = new ViewConfig("airports_countries");
+    private static final ViewConfig LANGUAGES_COUNTRIES = new ViewConfig("languages_countries");
+    public static final List<ViewConfig> VIEW_CONFIGS = List.of(ADDRESS_COUNTRIES, AIRPORTS_COUNTRIES, LANGUAGES_COUNTRIES);
+
     /**
      * <p>
      * Loads spec data on a local ES server.
@@ -402,14 +407,20 @@ private static void loadDataSetIntoEs(
         Logger logger = LogManager.getLogger(CsvTestsDataLoader.class);
 
         Set<String> loadedDatasets = new HashSet<>();
+        logger.info("Loading test datasets");
         for (var dataset : availableDatasetsForEs(supportsIndexModeLookup, supportsSourceFieldMapping, inferenceEnabled)) {
             load(client, dataset, logger, indexCreator);
             loadedDatasets.add(dataset.indexName);
         }
         forceMerge(client, loadedDatasets, logger);
+        logger.info("Loading enrich policies");
         for (var policy : ENRICH_POLICIES) {
             loadEnrichPolicy(client, policy.policyName, policy.policyFileName, logger);
         }
+        logger.info("Loading views");
+        for (var view : VIEW_CONFIGS) {
+            loadView(client, view.viewName, view.viewFileName, logger);
+        }
     }
 
     public static void createInferenceEndpoints(RestClient client) throws IOException {
@@ -548,6 +559,7 @@ private static void deleteInferenceEndpoint(RestClient client, String inferenceI
     }
 
     private static void loadEnrichPolicy(RestClient client, String policyName, String policyFileName, Logger logger) throws IOException {
+        logger.info("Loading enrich policy [{}] from file [{}]", policyName, policyFileName);
         URL policyMapping = getResource("/" + policyFileName);
         String entity = readTextFile(policyMapping);
         Request request = new Request("PUT", "/_enrich/policy/" + policyName);
@@ -558,6 +570,16 @@ private static void loadEnrichPolicy(RestClient client, String policyName, Strin
         client.performRequest(request);
     }
 
+    private static void loadView(RestClient client, String viewName, String viewFilename, Logger logger) throws IOException {
+        logger.info("Loading view [{}] from file [{}]", viewName, viewFilename);
+        URL viewFile = getResource("/" + viewFilename);
+        String viewQuery = readTextFile(viewFile);
+        Request request = new Request("PUT", "/_query/view/" + viewName);
+        request.setJsonEntity("{\"query\":\"" + viewQuery.replace("\"", "\\\"").replace("\n", " ") + "\"}");
+        Response response = client.performRequest(request);
+        logger.info("View creation response: {}", response.getStatusLine());
+    }
+
     private static URL getResource(String name) {
         URL result = CsvTestsDataLoader.class.getResource(name);
         if (result == null) {
@@ -567,6 +589,7 @@ private static URL getResource(String name) {
     }
 
     private static void load(RestClient client, TestDataset dataset, Logger logger, IndexCreator indexCreator) throws IOException {
+        logger.info("Loading dataset [{}] into ES index [{}]", dataset.dataFileName, dataset.indexName);
         URL mapping = getResource("/" + dataset.mappingFileName);
         Settings indexSettings = dataset.readSettingsFile();
         indexCreator.createIndex(client, dataset.indexName, readMappingFile(mapping, dataset.typeMapping), indexSettings);
@@ -933,6 +956,12 @@ private Settings readSettingsFile() throws IOException {
         }
     }
 
+    public record ViewConfig(String viewName, String viewFileName) {
+        public ViewConfig(String viewName) {
+            this(viewName, "views/" + viewName + ".esql");
+        }
+    }
+
     public record EnrichConfig(String policyName, String policyFileName) {}
 
     private interface IndexCreator {

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

@@ -0,0 +1,88 @@
+addressCountries
+FROM address_countries
+;
+
+count:long | country:keyword
+1          | Netherlands
+1          | United States of America
+1          | Japan
+;
+
+airports
+FROM airports
+| WHERE country IS NOT NULL
+| STATS count=COUNT() BY country
+| SORT count DESC
+| WHERE count > 10
+;
+
+count:long | country:keyword
+ 129        | United States
+ 50         | India
+ 45         | Mexico
+ 41         | China
+ 37         | Canada
+ 31         | Brazil
+ 26         | Russia
+ 17         | United Kingdom
+ 17         | Australia
+ 13         | Argentina
+ 12         | France
+ 12         | Indonesia
+ 11         | Nigeria
+ 11         | Italy
+ 11         | Germany
+;
+
+airportsCountries
+FROM airports_countries
+;
+
+count:long | country:keyword
+129        | United States
+50         | India
+45         | Mexico
+41         | China
+37         | Canada
+31         | Brazil
+26         | Russia
+17         | United Kingdom
+17         | Australia
+13         | Argentina
+12         | France
+12         | Indonesia
+11         | Nigeria
+11         | Italy
+11         | Germany
+;
+
+languages
+FROM languages_lookup_non_unique_key
+| MV_EXPAND country
+| WHERE country IS NOT NULL AND country NOT LIKE ("*-Land*")
+| STATS count=COUNT() BY country
+;
+
+count:long | country:text
+ 1         | Canada
+ 1         | United Kingdom
+ 1         | United States of America
+ 1         | Germany
+ 1         | Austria
+ 1         | Switzerland
+ 1         | Atlantis
+;
+
+languagesCountries
+FROM languages_countries
+;
+
+count:long | country:text
+ 1         | Canada
+ 1         | United Kingdom
+ 1         | United States of America
+ 1         | Germany
+ 1         | Austria
+ 1         | Switzerland
+ 1         | Atlantis
+;

x-pack/plugin/esql/qa/testFixtures/src/main/resources/views/address_countries.esql

@@ -0,0 +1,3 @@
+FROM addresses
+| RENAME city.country.name AS country
+| STATS count=COUNT() BY country

x-pack/plugin/esql/qa/testFixtures/src/main/resources/views/airports_countries.esql

@@ -0,0 +1,5 @@
+FROM airports
+| WHERE country IS NOT NULL
+| STATS count=COUNT() BY country
+| SORT count DESC
+| WHERE count > 10

x-pack/plugin/esql/qa/testFixtures/src/main/resources/views/languages_countries.esql

@@ -0,0 +1,4 @@
+FROM languages_lookup_non_unique_key
+| MV_EXPAND country
+| WHERE country IS NOT NULL AND country NOT LIKE ("*-Land*")
+| STATS count=COUNT() BY country

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

@@ -19,13 +19,13 @@
 import java.io.IOException;
 import java.util.List;
 
-import static org.elasticsearch.rest.RestRequest.Method.POST;
+import static org.elasticsearch.rest.RestRequest.Method.PUT;
 
 @ServerlessScope(Scope.PUBLIC)
 public class RestPutViewAction extends BaseRestHandler {
     @Override
     public List<Route> routes() {
-        return List.of(new Route(POST, "/_query/view/{name}"));
+        return List.of(new Route(PUT, "/_query/view/{name}"));
     }
 
     @Override