Add a multi-node test to sql (elastic/x-pack-elasticsearch#2136)

SQL relies on being able to fetch information about fields from
the cluster state and it'd be disasterous if that information
wasn't available. This should catch that.

Original commit: elastic/x-pack-elasticsearch@1a62747332

dev-tools/ci

@@ -53,6 +53,7 @@ case $key in
             "-psql"
             "check"
             ":x-pack-elasticsearch:plugin:precommit"
+            ":x-pack-elasticsearch:qa:sql-multinode:check"
             ":x-pack-elasticsearch:qa:sql-no-security:check"
             ":x-pack-elasticsearch:qa:sql-security:check"
             "-xforbiddenPatterns"

qa/sql-multinode/build.gradle

@@ -0,0 +1,21 @@
+import org.elasticsearch.gradle.test.RunTask
+
+apply plugin: 'elasticsearch.standalone-rest-test'
+apply plugin: 'elasticsearch.rest-test'
+
+dependencies {
+  testCompile project(path: ':x-pack-elasticsearch:plugin', configuration: 'runtime')
+  testCompile project(path: ':x-pack-elasticsearch:plugin', configuration: 'testArtifacts')
+  testCompile project(path: ':modules:reindex')
+}
+
+// NOCOMMIT we should try this on multiple nodes
+
+integTestCluster {
+  distribution = 'zip' // NOCOMMIT make double sure we want all the modules
+  numNodes = 2
+  plugin ':x-pack-elasticsearch:plugin'
+  setting 'xpack.ml.enabled', 'false'
+  setting 'xpack.monitoring.enabled', 'false'
+  setting 'xpack.security.enabled', 'false'
+}

qa/sql-multinode/src/test/java/org/elasticsearch/xpack/sql/SqlMultinodeIT.java

@@ -0,0 +1,112 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.sql;
+
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.client.http.HttpHost;
+import org.elasticsearch.client.http.entity.ContentType;
+import org.elasticsearch.client.http.entity.StringEntity;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentHelper;
+import org.elasticsearch.common.xcontent.json.JsonXContent;
+import org.elasticsearch.test.NotEqualMessageBuilder;
+import org.elasticsearch.test.rest.ESRestTestCase;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.UnsupportedCharsetException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static java.util.Collections.emptyMap;
+import static java.util.Collections.singletonList;
+import static java.util.Collections.singletonMap;
+
+public class SqlMultinodeIT extends ESRestTestCase {
+    /**
+     * Tests count of index run across multiple nodes.
+     */
+    public void testIndexSpread() throws IOException {
+        int documents = between(10, 100);
+        createTestData(documents);
+        assertCount(client(), documents);
+    }
+
+    /**
+     * Tests count against index on a node that doesn't have any shards of the index.
+     */
+    public void testIndexOnWrongNode() throws IOException {
+        HttpHost firstHost = getClusterHosts().get(0);
+        String firstHostName = null;
+
+        String match = firstHost.getHostName() + ":" + firstHost.getPort();
+        Map<String, Object> nodesInfo = responseToMap(client().performRequest("GET", "/_nodes"));
+        @SuppressWarnings("unchecked")
+        Map<String, Object> nodes = (Map<String, Object>) nodesInfo.get("nodes");
+        for (Map.Entry<String, Object> node : nodes.entrySet()) {
+            String name = node.getKey();
+            Map<?, ?> nodeEntries = (Map<?, ?>) node.getValue();
+            Map<?, ?> http = (Map<?, ?>) nodeEntries.get("http");
+            List<?> boundAddress = (List<?>) http.get("bound_address");
+            if (boundAddress.contains(match)) {
+                firstHostName = name;
+                break;
+            }
+        }
+        assertNotNull("Didn't find first host among published addresses", firstHostName);
+
+        XContentBuilder index = JsonXContent.contentBuilder().prettyPrint().startObject();
+        index.startObject("settings"); {
+            index.field("routing.allocation.exclude._name", firstHostName);
+        }
+        index.endObject();
+        index.endObject();
+        client().performRequest("PUT", "/test", emptyMap(), new StringEntity(index.string(), ContentType.APPLICATION_JSON));
+        int documents = between(10, 100);
+        createTestData(documents);
+
+        try (RestClient firstNodeClient = buildClient(restClientSettings(), new HttpHost[] {firstHost})) {
+            assertCount(firstNodeClient, documents);
+        }
+    }
+
+    private void createTestData(int documents) throws UnsupportedCharsetException, IOException {
+        StringBuilder bulk = new StringBuilder();
+        for (int i = 0; i < documents; i++) {
+            int a = 3 * i;
+            int b = a + 1;
+            int c = b + 1;
+            bulk.append("{\"index\":{\"_id\":\"" + i + "\"}\n");
+            bulk.append("{\"a\": " + a + ", \"b\": " + b + ", \"c\": " + c + "}\n");
+        }
+        client().performRequest("PUT", "/test/test/_bulk", singletonMap("refresh", "true"),
+                new StringEntity(bulk.toString(), ContentType.APPLICATION_JSON));
+    }
+
+    private Map<String, Object> responseToMap(Response response) throws IOException {
+        try (InputStream content = response.getEntity().getContent()) {
+            return XContentHelper.convertToMap(JsonXContent.jsonXContent, content, false);
+        }
+    }
+
+    private void assertCount(RestClient client, int count) throws IOException {
+        Map<String, Object> expected = new HashMap<>();
+        expected.put("columns", singletonMap("COUNT(1)", singletonMap("type", "long")));
+        expected.put("rows", singletonList(singletonMap("COUNT(1)", count)));
+        expected.put("size", 1);
+
+        Map<String, Object> actual = responseToMap(client.performRequest("POST", "/_sql", emptyMap(),
+                new StringEntity("{\"query\": \"SELECT COUNT(*) FROM test.test\"}", ContentType.APPLICATION_JSON)));
+
+        if (false == expected.equals(actual)) {
+            NotEqualMessageBuilder message = new NotEqualMessageBuilder();
+            message.compareMaps(actual, expected);
+            fail("Response does not match:\n" + message.toString());
+        }
+    }
+}

sql/server/src/main/java/org/elasticsearch/xpack/sql/expression/function/AbstractFunctionRegistry.java

@@ -155,8 +155,8 @@ abstract class AbstractFunctionRegistry implements FunctionRegistry {
                 if (timezoneAware) {
                     args.add(settings.timeZone());
                 }
-            } 
-            return (Function) info.ctr.newInstance(args);
+            }
+            return (Function) info.ctr.newInstance(args.toArray());
         } catch (InstantiationException | IllegalAccessException | IllegalArgumentException | InvocationTargetException ex) {
             throw new SqlIllegalArgumentException(ex, "Cannot create instance of function %s", ur.name());
         }