SQL: binary communication implementation for drivers and the CLI

x-pack/plugin/sql/jdbc/src/main/java/org/elasticsearch/xpack/sql/jdbc/JdbcConfiguration.java

@@ -60,10 +60,13 @@ public class JdbcConfiguration extends ConnectionConfiguration {
     static final String INDEX_INCLUDE_FROZEN = "index.include.frozen";
     static final String INDEX_INCLUDE_FROZEN_DEFAULT = "false";
 
+    static final String BINARY_RESPONSE = "binary.response";
+    static final String BINARY_RESPONSE_DEFAULT = "true";
+
 
     // options that don't change at runtime
     private static final Set<String> OPTION_NAMES = new LinkedHashSet<>(
-            Arrays.asList(TIME_ZONE, FIELD_MULTI_VALUE_LENIENCY, INDEX_INCLUDE_FROZEN, DEBUG, DEBUG_OUTPUT));
+            Arrays.asList(TIME_ZONE, FIELD_MULTI_VALUE_LENIENCY, INDEX_INCLUDE_FROZEN, DEBUG, DEBUG_OUTPUT, BINARY_RESPONSE));
 
     static {
         // trigger version initialization
@@ -81,6 +84,7 @@ public class JdbcConfiguration extends ConnectionConfiguration {
     private ZoneId zoneId;
     private boolean fieldMultiValueLeniency;
     private boolean includeFrozen;
+    private boolean binaryResponse;
 
     public static JdbcConfiguration create(String u, Properties props, int loginTimeoutSeconds) throws JdbcSQLException {
         URI uri = parseUrl(u);
@@ -165,6 +169,8 @@ private JdbcConfiguration(URI baseURI, String u, Properties props) throws JdbcSQ
                 props.getProperty(FIELD_MULTI_VALUE_LENIENCY, FIELD_MULTI_VALUE_LENIENCY_DEFAULT), Boolean::parseBoolean);
         this.includeFrozen = parseValue(INDEX_INCLUDE_FROZEN, props.getProperty(INDEX_INCLUDE_FROZEN, INDEX_INCLUDE_FROZEN_DEFAULT),
                 Boolean::parseBoolean);
+        this.binaryResponse = parseValue(BINARY_RESPONSE, props.getProperty(BINARY_RESPONSE, BINARY_RESPONSE_DEFAULT),
+                Boolean::parseBoolean);
     }
 
     @Override
@@ -195,6 +201,10 @@ public boolean fieldMultiValueLeniency() {
     public boolean indexIncludeFrozen() {
         return includeFrozen;
     }
+
+    public boolean binaryResponse() {
+        return binaryResponse;
+    }
 
     public static boolean canAccept(String url) {
         return (StringUtils.hasText(url) && url.trim().startsWith(JdbcConfiguration.URL_PREFIX));

x-pack/plugin/sql/jdbc/src/main/java/org/elasticsearch/xpack/sql/jdbc/JdbcHttpClient.java

@@ -64,7 +64,8 @@ Cursor query(String sql, List<SqlTypedParamValue> params, RequestMeta meta) thro
                 null,
                 new RequestInfo(Mode.JDBC),
                 conCfg.fieldMultiValueLeniency(),
-                conCfg.indexIncludeFrozen());
+                conCfg.indexIncludeFrozen(),
+                conCfg.binaryResponse());
         SqlQueryResponse response = httpClient.query(sqlRequest);
         return new DefaultCursor(this, response.cursor(), toJdbcColumnInfo(response.columns()), response.rows(), meta);
     }

x-pack/plugin/sql/qa/multi-node/src/test/java/org/elasticsearch/xpack/sql/qa/multi_node/RestSqlMultinodeIT.java

@@ -11,13 +11,11 @@
 import org.elasticsearch.client.RestClient;
 import org.elasticsearch.common.Strings;
 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.sql.JDBCType;
 import java.util.HashMap;
@@ -27,6 +25,7 @@
 import static java.util.Collections.singletonList;
 import static org.elasticsearch.xpack.sql.qa.rest.BaseRestSqlTestCase.mode;
 import static org.elasticsearch.xpack.sql.qa.rest.BaseRestSqlTestCase.randomMode;
+import static org.elasticsearch.xpack.sql.qa.rest.BaseRestSqlTestCase.toMap;
 import static org.elasticsearch.xpack.sql.qa.rest.RestSqlTestCase.SQL_QUERY_REST_ENDPOINT;
 import static org.elasticsearch.xpack.sql.qa.rest.RestSqlTestCase.columnInfo;
 
@@ -101,9 +100,7 @@ private void createTestData(int documents) throws UnsupportedCharsetException, I
     }
 
     private Map<String, Object> responseToMap(Response response) throws IOException {
-        try (InputStream content = response.getEntity().getContent()) {
-            return XContentHelper.convertToMap(JsonXContent.jsonXContent, content, false);
-        }
+        return toMap(response, "plain");
     }
 
     private void assertCount(RestClient client, int count) throws IOException {
@@ -114,7 +111,7 @@ private void assertCount(RestClient client, int count) throws IOException {
 
         Request request = new Request("POST", SQL_QUERY_REST_ENDPOINT);
         request.setJsonEntity("{\"query\": \"SELECT COUNT(*) FROM test\"" + mode(mode) + "}");
-        Map<String, Object> actual = responseToMap(client.performRequest(request));
+        Map<String, Object> actual = toMap(client.performRequest(request), mode);
 
         if (false == expected.equals(actual)) {
             NotEqualMessageBuilder message = new NotEqualMessageBuilder();

x-pack/plugin/sql/qa/multi-node/src/test/java/org/elasticsearch/xpack/sql/qa/multi_node/SqlProtocolIT.java

@@ -0,0 +1,12 @@
+/*
+ * 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.qa.multi_node;
+
+import org.elasticsearch.xpack.sql.qa.SqlProtocolTestCase;
+
+public class SqlProtocolIT extends SqlProtocolTestCase {
+}

x-pack/plugin/sql/qa/security/src/test/java/org/elasticsearch/xpack/sql/qa/security/RestSqlSecurityIT.java

@@ -14,6 +14,7 @@
 import org.elasticsearch.client.ResponseException;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.xcontent.XContentHelper;
+import org.elasticsearch.common.xcontent.cbor.CborXContent;
 import org.elasticsearch.common.xcontent.json.JsonXContent;
 import org.elasticsearch.test.NotEqualMessageBuilder;
 import org.hamcrest.Matcher;
@@ -70,10 +71,10 @@ public void expectScrollMatchesAdmin(String adminSql, String user, String userSq
             String mode = randomMode();
             Map<String, Object> adminResponse = runSql(null,
                     new StringEntity("{\"query\": \"" + adminSql + "\", \"fetch_size\": 1" + mode(mode) + "}",
-                            ContentType.APPLICATION_JSON));
+                            ContentType.APPLICATION_JSON), mode);
             Map<String, Object> otherResponse = runSql(user,
                     new StringEntity("{\"query\": \"" + adminSql + "\", \"fetch_size\": 1" + mode(mode) + "}",
-                            ContentType.APPLICATION_JSON));
+                            ContentType.APPLICATION_JSON), mode);
 
             String adminCursor = (String) adminResponse.remove("cursor");
             String otherCursor = (String) otherResponse.remove("cursor");
@@ -82,9 +83,9 @@ public void expectScrollMatchesAdmin(String adminSql, String user, String userSq
             assertResponse(adminResponse, otherResponse);
             while (true) {
                 adminResponse = runSql(null,
-                        new StringEntity("{\"cursor\": \"" + adminCursor + "\"" + mode(mode) + "}", ContentType.APPLICATION_JSON));
+                        new StringEntity("{\"cursor\": \"" + adminCursor + "\"" + mode(mode) + "}", ContentType.APPLICATION_JSON), mode);
                 otherResponse = runSql(user,
-                        new StringEntity("{\"cursor\": \"" + otherCursor + "\"" + mode(mode) + "}", ContentType.APPLICATION_JSON));
+                        new StringEntity("{\"cursor\": \"" + otherCursor + "\"" + mode(mode) + "}", ContentType.APPLICATION_JSON), mode);
                 adminCursor = (String) adminResponse.remove("cursor");
                 otherCursor = (String) otherResponse.remove("cursor");
                 assertResponse(adminResponse, otherResponse);
@@ -179,18 +180,18 @@ public void checkNoMonitorMain(String user) throws Exception {
         }
 
         private static Map<String, Object> runSql(@Nullable String asUser, String mode, String sql) throws IOException {
-            return runSql(asUser, new StringEntity("{\"query\": \"" + sql + "\"" + mode(mode) + "}", ContentType.APPLICATION_JSON));
+            return runSql(asUser, new StringEntity("{\"query\": \"" + sql + "\"" + mode(mode) + "}", ContentType.APPLICATION_JSON), mode);
         }
 
-        private static Map<String, Object> runSql(@Nullable String asUser, HttpEntity entity) throws IOException {
+        private static Map<String, Object> runSql(@Nullable String asUser, HttpEntity entity, String mode) throws IOException {
             Request request = new Request("POST", SQL_QUERY_REST_ENDPOINT);
             if (asUser != null) {
                 RequestOptions.Builder options = request.getOptions().toBuilder();
                 options.addHeader("es-security-runas-user", asUser);
                 request.setOptions(options);
             }
             request.setEntity(entity);
-            return toMap(client().performRequest(request));
+            return toMap(client().performRequest(request), mode);
         }
 
         private static void assertResponse(Map<String, Object> expected, Map<String, Object> actual) {
@@ -201,9 +202,13 @@ private static void assertResponse(Map<String, Object> expected, Map<String, Obj
             }
         }
 
-        private static Map<String, Object> toMap(Response response) throws IOException {
+        private static Map<String, Object> toMap(Response response, String mode) throws IOException {
             try (InputStream content = response.getEntity().getContent()) {
-                return XContentHelper.convertToMap(JsonXContent.jsonXContent, content, false);
+                if (mode.equalsIgnoreCase("jdbc")) {
+                    return XContentHelper.convertToMap(CborXContent.cborXContent, content, false);
+                } else {
+                    return XContentHelper.convertToMap(JsonXContent.jsonXContent, content, false);
+                }
             }
         }
     }
@@ -226,15 +231,17 @@ protected AuditLogAsserter createAuditLogAsserter() {
     public void testHijackScrollFails() throws Exception {
         createUser("full_access", "rest_minimal");
 
+        String mode = randomMode();
         Map<String, Object> adminResponse = RestActions.runSql(null,
-                new StringEntity("{\"query\": \"SELECT * FROM test\", \"fetch_size\": 1" + mode(randomMode()) + "}",
-                        ContentType.APPLICATION_JSON));
+                new StringEntity("{\"query\": \"SELECT * FROM test\", \"fetch_size\": 1" + mode(mode) + "}",
+                        ContentType.APPLICATION_JSON), mode);
 
         String cursor = (String) adminResponse.remove("cursor");
         assertNotNull(cursor);
 
+        final String m = randomMode();
         ResponseException e = expectThrows(ResponseException.class, () -> RestActions.runSql("full_access",
-                new StringEntity("{\"cursor\":\"" + cursor + "\"" + mode(randomMode()) + "}", ContentType.APPLICATION_JSON)));
+                new StringEntity("{\"cursor\":\"" + cursor + "\"" + mode(m) + "}", ContentType.APPLICATION_JSON), m));
         // TODO return a better error message for bad scrolls
         assertThat(e.getMessage(), containsString("No search context found for id"));
         assertEquals(404, e.getResponse().getStatusLine().getStatusCode());

x-pack/plugin/sql/qa/security/src/test/java/org/elasticsearch/xpack/sql/qa/security/UserFunctionIT.java

@@ -6,16 +6,13 @@
 
 package org.elasticsearch.xpack.sql.qa.security;
 
-import org.apache.http.HttpEntity;
 import org.apache.http.entity.ContentType;
 import org.apache.http.entity.StringEntity;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.RequestOptions;
-import org.elasticsearch.client.Response;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.settings.Settings;
 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;
@@ -25,7 +22,6 @@
 import org.junit.rules.TestName;
 
 import java.io.IOException;
-import java.io.InputStream;
 import java.sql.JDBCType;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -36,6 +32,7 @@
 
 import static org.elasticsearch.xpack.sql.qa.rest.BaseRestSqlTestCase.mode;
 import static org.elasticsearch.xpack.sql.qa.rest.BaseRestSqlTestCase.randomMode;
+import static org.elasticsearch.xpack.sql.qa.rest.BaseRestSqlTestCase.toMap;
 import static org.elasticsearch.xpack.sql.qa.rest.RestSqlTestCase.SQL_QUERY_REST_ENDPOINT;
 import static org.elasticsearch.xpack.sql.qa.rest.RestSqlTestCase.columnInfo;
 
@@ -174,18 +171,14 @@ private void deleteUser(String name) throws IOException {
     }
 
     private Map<String, Object> runSql(String asUser, String mode, String sql) throws IOException {
-        return runSql(asUser, new StringEntity("{\"query\": \"" + sql + "\"" + mode(mode) + "}", ContentType.APPLICATION_JSON));
-    }
-
-    private Map<String, Object> runSql(String asUser, HttpEntity entity) throws IOException {
         Request request = new Request("POST", SQL_QUERY_REST_ENDPOINT);
         if (asUser != null) {
             RequestOptions.Builder options = request.getOptions().toBuilder();
             options.addHeader("es-security-runas-user", asUser);
             request.setOptions(options);
         }
-        request.setEntity(entity);
-        return toMap(client().performRequest(request));
+        request.setEntity(new StringEntity("{\"query\": \"" + sql + "\"" + mode(mode) + "}", ContentType.APPLICATION_JSON));
+        return toMap(client().performRequest(request), mode);
     }
 
     private void assertResponse(Map<String, Object> expected, Map<String, Object> actual) {
@@ -195,12 +188,6 @@ private void assertResponse(Map<String, Object> expected, Map<String, Object> ac
             fail("Response does not match:\n" + message.toString());
         }
     }
-
-    private static Map<String, Object> toMap(Response response) throws IOException {
-        try (InputStream content = response.getEntity().getContent()) {
-            return XContentHelper.convertToMap(JsonXContent.jsonXContent, content, false);
-        }
-    }
 
     private void index(String... docs) throws IOException {
         Request request = new Request("POST", "/test/_bulk");

x-pack/plugin/sql/qa/single-node/src/test/java/org/elasticsearch/xpack/sql/qa/single_node/RestSqlIT.java

@@ -5,8 +5,6 @@
  */
 package org.elasticsearch.xpack.sql.qa.single_node;
 
-import org.apache.http.entity.ContentType;
-import org.apache.http.entity.StringEntity;
 import org.elasticsearch.xpack.sql.qa.rest.RestSqlTestCase;
 
 import java.io.IOException;
@@ -22,28 +20,22 @@ public class RestSqlIT extends RestSqlTestCase {
 
     public void testErrorMessageForTranslatingQueryWithWhereEvaluatingToFalse() throws IOException {
         index("{\"foo\":1}");
-        expectBadRequest(() -> runSql(
-            new StringEntity("{\"query\":\"SELECT * FROM test WHERE foo = 1 AND foo = 2\"}",
-                ContentType.APPLICATION_JSON), "/translate/"),
+        expectBadRequest(() -> runTranslateSql("{\"query\":\"SELECT * FROM test WHERE foo = 1 AND foo = 2\"}"),
             containsString("Cannot generate a query DSL for an SQL query that either its WHERE clause evaluates " +
                 "to FALSE or doesn't operate on a table (missing a FROM clause), sql statement: " +
                 "[SELECT * FROM test WHERE foo = 1 AND foo = 2]"));
     }
 
     public void testErrorMessageForTranslatingQueryWithLocalExecution() throws IOException {
         index("{\"foo\":1}");
-        expectBadRequest(() -> runSql(
-            new StringEntity("{\"query\":\"SELECT SIN(PI())\"}",
-                ContentType.APPLICATION_JSON), "/translate/"),
+        expectBadRequest(() -> runTranslateSql("{\"query\":\"SELECT SIN(PI())\"}"),
             containsString("Cannot generate a query DSL for an SQL query that either its WHERE clause evaluates " +
                 "to FALSE or doesn't operate on a table (missing a FROM clause), sql statement: [SELECT SIN(PI())]"));
     }
 
     public void testErrorMessageForTranslatingSQLCommandStatement() throws IOException {
         index("{\"foo\":1}");
-        expectBadRequest(() -> runSql(
-            new StringEntity("{\"query\":\"SHOW FUNCTIONS\"}",
-                ContentType.APPLICATION_JSON), "/translate/"),
+        expectBadRequest(() -> runTranslateSql("{\"query\":\"SHOW FUNCTIONS\"}"),
             containsString("Cannot generate a query DSL for a special SQL command " +
                 "(e.g.: DESCRIBE, SHOW), sql statement: [SHOW FUNCTIONS]"));
     }

x-pack/plugin/sql/qa/src/main/java/org/elasticsearch/xpack/sql/qa/SqlProtocolTestCase.java

@@ -195,6 +195,18 @@ private Map<String, Object> runSql(String mode, String sql, boolean columnar) th
             requestContent = new StringBuilder(requestContent)
                     .insert(requestContent.length() - 1, ",\"columnar\":" + columnar).toString();
         }
+
+        // randomize binary response enforcement for drivers (ODBC/JDBC) and CLI
+        boolean binaryResponse = randomBoolean();
+        Mode m = Mode.fromString(mode);
+        if (randomBoolean()) {
+            // set it explicitly or leave the default (null) as is
+            requestContent = new StringBuilder(requestContent)
+                    .insert(requestContent.length() - 1, ",\"binary_response\":" + binaryResponse).toString();
+            binaryResponse = ((Mode.isDriver(m) && binaryResponse == true) || m == Mode.CLI);
+        } else {
+            binaryResponse = Mode.isDriver(m) || m == Mode.CLI;
+        }
 
         // send the query either as body or as request parameter
         if (randomBoolean()) {
@@ -210,6 +222,9 @@ private Map<String, Object> runSql(String mode, String sql, boolean columnar) th
 
         Response response = client().performRequest(request);
         try (InputStream content = response.getEntity().getContent()) {
+            if (binaryResponse == true) {
+                return XContentHelper.convertToMap(CborXContent.cborXContent, content, false);
+            }
             switch(format) {
                 case "cbor": {
                     return XContentHelper.convertToMap(CborXContent.cborXContent, content, false);

x-pack/plugin/sql/qa/src/main/java/org/elasticsearch/xpack/sql/qa/rest/BaseRestSqlTestCase.java

@@ -7,11 +7,18 @@
 package org.elasticsearch.xpack.sql.qa.rest;
 
 import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
 import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.xcontent.XContentHelper;
+import org.elasticsearch.common.xcontent.cbor.CborXContent;
+import org.elasticsearch.common.xcontent.json.JsonXContent;
 import org.elasticsearch.test.rest.ESRestTestCase;
+import org.elasticsearch.xpack.sql.proto.Mode;
 import org.elasticsearch.xpack.sql.proto.StringUtils;
 
 import java.io.IOException;
+import java.io.InputStream;
+import java.util.Map;
 
 public abstract class BaseRestSqlTestCase extends ESRestTestCase {
 
@@ -34,4 +41,27 @@ public static String mode(String mode) {
     public static String randomMode() {
         return randomFrom(StringUtils.EMPTY, "jdbc", "plain");
     }
+
+    /**
+     * CBOR XContent parser returns floating point numbers as Floats, while JSON parser as Doubles.
+     * To have the tests compare the correct data type, the floating point numbers types should be passed accordingly, to the comparators.
+     */
+    public static Number xContentDependentFloatingNumberValue(String mode, Number value) {
+        Mode m = Mode.fromString(mode);
+        if (Mode.isDriver(m) || m == Mode.CLI) {
+            return value.floatValue();
+        } else {
+            return value.doubleValue();
+        }
+    }
+
+    public static Map<String, Object> toMap(Response response, String mode) throws IOException {
+        try (InputStream content = response.getEntity().getContent()) {
+            if (Mode.fromString(mode) == Mode.JDBC) {
+                return XContentHelper.convertToMap(CborXContent.cborXContent, content, false);
+            } else {
+                return XContentHelper.convertToMap(JsonXContent.jsonXContent, content, false);
+            }
+        }
+    }
 }