SQL: Add support for plain text output to /_sql endpoint (elastic/x-pack-elasticsearch#3124)

The /_sql endpoint now returns the results in the text format by default. Structured formats are also supported using the format parameter or accept header similar to _cat endpoints.

Original commit: elastic/x-pack-elasticsearch@4353793b83

docs/en/sql/endpoints/sql-rest.asciidoc

@@ -5,9 +5,45 @@
 The SQL REST API accepts SQL in a JSON document, executes it,
 and returns the results. For example:
 
+
 [source,js]
 --------------------------------------------------
 POST /_sql
+{
+    "query": "SELECT * FROM library ORDER BY page_count DESC LIMIT 5"
+}
+--------------------------------------------------
+// CONSOLE
+// TEST[setup:library]
+
+Which returns:
+
+[source,text]
+--------------------------------------------------
+     author      |        name        |  page_count   | release_date
+-----------------+--------------------+---------------+---------------
+Peter F. Hamilton|Pandora's Star      |768            |1078185600000
+Vernor Vinge     |A Fire Upon the Deep|613            |707356800000
+Frank Herbert    |Dune                |604            |-144720000000
+Alastair Reynolds|Revelation Space    |585            |953078400000
+James S.A. Corey |Leviathan Wakes     |561            |1306972800000
+--------------------------------------------------
+// TESTRESPONSE[_cat]
+
+You can also choose to get results in a structured format by adding the `format` parameter. Currently supported formats:
+- text (default)
+- json
+- smile
+- yaml
+- cbor
+
+Alternatively you can set the Accept HTTP header to the appropriate media format.
+All formats above are supported, the GET parameter takes precedence over the header.
+
+
+[source,js]
+--------------------------------------------------
+POST /_sql?format=json
 {
     "query": "SELECT * FROM library ORDER BY page_count DESC",
     "fetch_size": 5
@@ -40,11 +76,12 @@ Which returns:
 --------------------------------------------------
 // TESTRESPONSE[s/sDXF1ZXJ5QW5kRmV0Y2gBAAAAAAAAAAEWWWdrRlVfSS1TbDYtcW9lc1FJNmlYdw==:BAFmBmF1dGhvcgFmBG5hbWUBZgpwYWdlX2NvdW50AWYMcmVsZWFzZV9kYXRl\+v\/\/\/w8=/$body.cursor/]
 
-You can continue to the next page by sending back the `cursor` field:
+You can continue to the next page by sending back the `cursor` field. In
+case of text format the cursor is returned as `Cursor` http header.
 
 [source,js]
 --------------------------------------------------
-POST /_sql
+POST /_sql?format=json
 {
     "cursor": "sDXF1ZXJ5QW5kRmV0Y2gBAAAAAAAAAAEWYUpOYklQMHhRUEtld3RsNnFtYU1hQQ==:BAFmBGRhdGUBZgVsaWtlcwFzB21lc3NhZ2UBZgR1c2Vy9f///w8="
 }
@@ -107,22 +144,13 @@ POST /_sql
 
 Which returns:
 
-[source,js]
+[source,text]
 --------------------------------------------------
-{
+    author     |                name                |  page_count   | release_date
-    "columns": [
+---------------+------------------------------------+---------------+---------------
-        {"name": "author",       "type": "keyword"},
+Douglas Adams  |The Hitchhiker's Guide to the Galaxy|180            |308534400000
-        {"name": "name",         "type": "keyword"},
-        {"name": "page_count",   "type": "short"},
-        {"name": "release_date", "type": "date"}
-    ],
-    "size": 1,
-    "rows": [
-        ["Douglas Adams",  "The Hitchhiker's Guide to the Galaxy",       180, 308534400000]
-    ]
-}
 --------------------------------------------------
-// TESTRESPONSE
+// TESTRESPONSE[_cat]
 
 [[sql-rest-fields]]
 In addition to the `query` and `cursor` fields, the request can

plugin/src/test/resources/rest-api-spec/api/xpack.sql.json

@@ -6,7 +6,12 @@
         "path": "/_sql",
         "paths": [ "/_sql" ],
         "parts": {},
-        "params": {}
+        "params": {
+          "format": {
+            "type" : "string",
+            "description" : "a short version of the Accept header, e.g. json, yaml"
+          }
+        }
       },
       "body": {
         "description" : "Use the `query` element to start a query. Use the `cursor` element to continue a query.",

plugin/src/test/resources/rest-api-spec/test/sql/sql.yml

@@ -19,6 +19,7 @@
 
   - do:
       xpack.sql:
+        format: json
         body:
           query: "SELECT * FROM test ORDER BY int asc"
   - match: { columns.0.name: int }
@@ -27,3 +28,16 @@
   - match: { rows.0.1: test1 }
   - match: { rows.1.0: 2 }
   - match: { rows.1.1: test2 }
+
+  - do:
+      xpack.sql:
+        format: text
+        body:
+          query: "SELECT * FROM test ORDER BY int asc"
+  - match:
+      $body: |
+               /^  \s+   int \s+  \| \s+  str \s+  \n
+                   ---------------\+---------------\n
+                   1      \s+     \|test1   \s+    \n
+                   2      \s+     \|test2   \s+    \n
+               $/

qa/sql/multinode/src/test/java/org/elasticsearch/xpack/qa/sql/multinode/RestSqlMultinodeIT.java

@@ -104,7 +104,7 @@ public class RestSqlMultinodeIT extends ESRestTestCase {
         expected.put("rows", singletonList(singletonList(count)));
         expected.put("size", 1);
 
-        Map<String, Object> actual = responseToMap(client.performRequest("POST", "/_sql", emptyMap(),
+        Map<String, Object> actual = responseToMap(client.performRequest("POST", "/_sql", singletonMap("format", "json"),
                 new StringEntity("{\"query\": \"SELECT COUNT(*) FROM test\"}", ContentType.APPLICATION_JSON)));
 
         if (false == expected.equals(actual)) {

qa/sql/security/src/test/java/org/elasticsearch/xpack/qa/sql/security/RestSqlSecurityIT.java

@@ -24,6 +24,7 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import static java.util.Collections.singletonMap;
 import static org.elasticsearch.xpack.qa.sql.rest.RestSqlTestCase.columnInfo;
 import static java.util.Collections.emptyMap;
 import static java.util.Collections.singletonList;
@@ -125,7 +126,7 @@ public class RestSqlSecurityIT extends SqlSecurityTestCase {
 
         private static Map<String, Object> runSql(@Nullable String asUser, HttpEntity entity) throws IOException {
             Header[] headers = asUser == null ? new Header[0] : new Header[] {new BasicHeader("es-security-runas-user", asUser)};
-            Response response = client().performRequest("POST", "/_sql", emptyMap(), entity, headers);
+            Response response = client().performRequest("POST", "/_sql", singletonMap("format", "json"), entity, headers);
             return toMap(response);
         }
 

qa/sql/src/main/java/org/elasticsearch/xpack/qa/sql/rest/RestSqlTestCase.java

@@ -9,6 +9,8 @@ import org.apache.http.entity.ContentType;
 import org.apache.http.entity.StringEntity;
 import org.elasticsearch.client.Response;
 import org.elasticsearch.client.ResponseException;
+import org.elasticsearch.common.collect.Tuple;
+import org.elasticsearch.common.io.Streams;
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.common.xcontent.json.JsonXContent;
 import org.elasticsearch.test.NotEqualMessageBuilder;
@@ -17,8 +19,11 @@ import org.hamcrest.Matcher;
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
 import java.util.HashMap;
+import java.util.Locale;
 import java.util.Map;
 
 import static java.util.Collections.emptyList;
@@ -99,7 +104,7 @@ public abstract class RestSqlTestCase extends ESRestTestCase {
         assertResponse(expected, runSql(new StringEntity("{\"cursor\":\"" + cursor + "\"}", ContentType.APPLICATION_JSON)));
     }
 
-    @AwaitsFix(bugUrl="https://github.com/elastic/x-pack-elasticsearch/issues/2074")
+    @AwaitsFix(bugUrl = "https://github.com/elastic/x-pack-elasticsearch/issues/2074")
     public void testTimeZone() throws IOException {
         StringBuilder bulk = new StringBuilder();
         bulk.append("{\"index\":{\"_id\":\"1\"}}\n");
@@ -147,29 +152,20 @@ public abstract class RestSqlTestCase extends ESRestTestCase {
         return runSql(suffix, new StringEntity("{\"query\":\"" + sql + "\"}", ContentType.APPLICATION_JSON));
     }
 
-    private Map<String, Object> runSql(String sql, String filter, String suffix) throws IOException {
-        return runSql(suffix, new StringEntity("{\"query\":\"" + sql + "\", \"filter\":" + filter + "}", ContentType.APPLICATION_JSON));
-    }
-
     private Map<String, Object> runSql(HttpEntity sql) throws IOException {
         return runSql("", sql);
     }
 
     private Map<String, Object> runSql(String suffix, HttpEntity sql) throws IOException {
-        Response response = client().performRequest("POST", "/_sql" + suffix, singletonMap("error_trace", "true"), sql);
+        Map<String, String> params = new HashMap<>();
+        params.put("error_trace", "true");
+        params.put("format", "json");
+        Response response = client().performRequest("POST", "/_sql" + suffix, params, sql);
         try (InputStream content = response.getEntity().getContent()) {
             return XContentHelper.convertToMap(JsonXContent.jsonXContent, content, false);
         }
     }
 
-    private void assertResponse(Map<String, Object> expected, Map<String, Object> actual) {
-        if (false == expected.equals(actual)) {
-            NotEqualMessageBuilder message = new NotEqualMessageBuilder();
-            message.compareMaps(actual, expected);
-            fail("Response does not match:\n" + message.toString());
-        }
-    }
-
     public void testBasicTranslateQuery() throws IOException {
         StringBuilder bulk = new StringBuilder();
         bulk.append("{\"index\":{\"_id\":\"1\"}}\n");
@@ -201,7 +197,8 @@ public abstract class RestSqlTestCase extends ESRestTestCase {
         expected.put("columns", singletonList(columnInfo("test", "text")));
         expected.put("rows", singletonList(singletonList("foo")));
         expected.put("size", 1);
-        assertResponse(expected, runSql("SELECT * FROM test", "{\"match\": {\"test\": \"foo\"}}", ""));
+        assertResponse(expected, runSql(new StringEntity("{\"query\":\"SELECT * FROM test\", \"filter\":{\"match\": {\"test\": \"foo\"}}}",
+                ContentType.APPLICATION_JSON)));
     }
 
     public void testBasicTranslateQueryWithFilter() throws IOException {
@@ -213,7 +210,10 @@ public abstract class RestSqlTestCase extends ESRestTestCase {
         client().performRequest("POST", "/test/test/_bulk", singletonMap("refresh", "true"),
                 new StringEntity(bulk.toString(), ContentType.APPLICATION_JSON));
 
-        Map<String, Object> response = runSql("SELECT * FROM test", "{\"match\": {\"test\": \"foo\"}}", "/translate/");
+        Map<String, Object> response = runSql("/translate/",
+                new StringEntity("{\"query\":\"SELECT * FROM test\", \"filter\":{\"match\": {\"test\": \"foo\"}}}",
+                        ContentType.APPLICATION_JSON));
+
         assertEquals(response.get("size"), 1000);
         @SuppressWarnings("unchecked")
         Map<String, Object> source = (Map<String, Object>) response.get("_source");
@@ -243,4 +243,79 @@ public abstract class RestSqlTestCase extends ESRestTestCase {
         assertEquals("foo", matchQuery.get("query"));
     }
 
+    public void testBasicQueryText() throws IOException {
+        StringBuilder bulk = new StringBuilder();
+        bulk.append("{\"index\":{\"_id\":\"1\"}}\n");
+        bulk.append("{\"test\":\"test\"}\n");
+        bulk.append("{\"index\":{\"_id\":\"2\"}}\n");
+        bulk.append("{\"test\":\"test\"}\n");
+        client().performRequest("POST", "/test/test/_bulk", singletonMap("refresh", "true"),
+                new StringEntity(bulk.toString(), ContentType.APPLICATION_JSON));
+        String expected =
+                "test           \n" +
+                "---------------\n" +
+                "test           \n" +
+                "test           \n";
+        Tuple<String, String> response = runSqlAsText("SELECT * FROM test");
+        logger.warn(expected);
+        logger.warn(response.v1());
+    }
+
+    public void testNextPageText() throws IOException {
+        StringBuilder bulk = new StringBuilder();
+        for (int i = 0; i < 20; i++) {
+            bulk.append("{\"index\":{\"_id\":\"" + i + "\"}}\n");
+            bulk.append("{\"text\":\"text" + i + "\", \"number\":" + i + "}\n");
+        }
+        client().performRequest("POST", "/test/test/_bulk", singletonMap("refresh", "true"),
+                new StringEntity(bulk.toString(), ContentType.APPLICATION_JSON));
+
+        String request = "{\"query\":\"SELECT text, number, number + 5 AS sum FROM test ORDER BY number\", \"fetch_size\":2}";
+
+        String cursor = null;
+        for (int i = 0; i < 20; i += 2) {
+            Tuple<String, String> response;
+            if (i == 0) {
+                response = runSqlAsText("", new StringEntity(request, ContentType.APPLICATION_JSON));
+            } else {
+                response = runSqlAsText("", new StringEntity("{\"cursor\":\"" + cursor + "\"}", ContentType.APPLICATION_JSON));
+            }
+
+            StringBuilder expected = new StringBuilder();
+            if (i == 0) {
+                expected.append("     text      |    number     |      sum      \n");
+                expected.append("---------------+---------------+---------------\n");
+            }
+            expected.append(String.format(Locale.ROOT, "%-15s|%-15d|%-15d\n", "text" + i, i, i + 5));
+            expected.append(String.format(Locale.ROOT, "%-15s|%-15d|%-15d\n", "text" + (i + 1), i + 1, i + 6));
+            cursor = response.v2();
+            assertEquals(expected.toString(), response.v1());
+            assertNotNull(cursor);
+        }
+        Map<String, Object> expected = new HashMap<>();
+        expected.put("size", 0);
+        expected.put("rows", emptyList());
+        assertResponse(expected, runSql(new StringEntity("{\"cursor\":\"" + cursor + "\"}", ContentType.APPLICATION_JSON)));
+    }
+
+    private Tuple<String, String> runSqlAsText(String sql) throws IOException {
+        return runSqlAsText("", new StringEntity("{\"query\":\"" + sql + "\"}", ContentType.APPLICATION_JSON));
+    }
+
+    private Tuple<String, String> runSqlAsText(String suffix, HttpEntity sql) throws IOException {
+        Response  response = client().performRequest("POST", "/_sql" + suffix, singletonMap("error_trace", "true"), sql);
+        return new Tuple<>(
+                Streams.copyToString(new InputStreamReader(response.getEntity().getContent(), StandardCharsets.UTF_8)),
+                response.getHeader("Cursor")
+        );
+    }
+
+    private void assertResponse(Map<String, Object> expected, Map<String, Object> actual) {
+        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/plugin/CliFormatterCursor.java

@@ -0,0 +1,56 @@
+/*
+ * 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.plugin;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.xpack.sql.session.Configuration;
+import org.elasticsearch.xpack.sql.session.Cursor;
+import org.elasticsearch.xpack.sql.session.RowSet;
+
+import java.io.IOException;
+
+/**
+ * The cursor that wraps all necessary information for textual representation of the result table
+ */
+public class CliFormatterCursor implements Cursor {
+    public static final String NAME = "f";
+
+    private Cursor delegate;
+    private CliFormatter formatter;
+
+    public CliFormatterCursor(Cursor delegate, CliFormatter formatter) {
+        this.delegate = delegate;
+        this.formatter = formatter;
+    }
+
+    public CliFormatterCursor(StreamInput in) throws IOException {
+        delegate = in.readNamedWriteable(Cursor.class);
+        formatter = new CliFormatter(in);
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeNamedWriteable(delegate);
+        formatter.writeTo(out);
+    }
+    public CliFormatter getCliFormatter() {
+        return formatter;
+    }
+
+    @Override
+    public void nextPage(Configuration cfg, Client client, ActionListener<RowSet> listener) {
+        delegate.nextPage(cfg, client, listener);
+    }
+
+    @Override
+    public String getWriteableName() {
+        return NAME;
+    }
+
+}

sql/server/src/main/java/org/elasticsearch/xpack/sql/plugin/sql/rest/RestSqlAction.java

@@ -5,18 +5,28 @@
  */
 package org.elasticsearch.xpack.sql.plugin.sql.rest;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.rest.BaseRestHandler;
+import org.elasticsearch.rest.BytesRestResponse;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
+import org.elasticsearch.rest.RestResponse;
+import org.elasticsearch.rest.RestStatus;
+import org.elasticsearch.rest.action.RestResponseListener;
 import org.elasticsearch.rest.action.RestToXContentListener;
+import org.elasticsearch.xpack.sql.plugin.CliFormatter;
+import org.elasticsearch.xpack.sql.plugin.CliFormatterCursor;
 import org.elasticsearch.xpack.sql.plugin.sql.action.SqlAction;
 import org.elasticsearch.xpack.sql.plugin.sql.action.SqlRequest;
 import org.elasticsearch.xpack.sql.plugin.sql.action.SqlResponse;
+import org.elasticsearch.xpack.sql.session.Cursor;
 
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 import static org.elasticsearch.rest.RestRequest.Method.POST;
@@ -35,8 +45,46 @@ public class RestSqlAction extends BaseRestHandler {
             sqlRequest = SqlRequest.PARSER.apply(parser, null);
         }
 
-        return channel -> client.executeLocally(
+        XContentType xContentType = XContentType.fromMediaTypeOrFormat(request.param("format", request.header("Accept")));
-                SqlAction.INSTANCE, sqlRequest, new RestToXContentListener<SqlResponse>(channel));
+        if (xContentType != null) {
+            // The client expects us to send back results in a XContent format
+            return channel -> client.executeLocally(SqlAction.INSTANCE, sqlRequest, new RestToXContentListener<>(channel));
+        }
+        // The client accepts plain text
+        long startNanos = System.nanoTime();
+        return channel -> client.execute(SqlAction.INSTANCE, sqlRequest, new RestResponseListener<SqlResponse>(channel) {
+            @Override
+            public RestResponse buildResponse(SqlResponse response) throws Exception {
+                final String data;
+                final CliFormatter formatter;
+                if (sqlRequest.cursor() != Cursor.EMPTY) {
+                    formatter = ((CliFormatterCursor) sqlRequest.cursor()).getCliFormatter();
+                    data = formatter.formatWithoutHeader(response);
+                } else {
+                    formatter = new CliFormatter(response);
+                    data = formatter.formatWithHeader(response);
+                }
+
+                final Cursor responseCursor;
+                if (response.cursor() == Cursor.EMPTY) {
+                    responseCursor = Cursor.EMPTY;
+                } else {
+                    responseCursor = new CliFormatterCursor(response.cursor(), formatter);
+                }
+                return buildTextResponse(responseCursor, System.nanoTime() - startNanos, data);
+            }
+        });
+    }
+
+    private RestResponse buildTextResponse(Cursor responseCursor, long tookNanos, String data)
+            throws IOException {
+        RestResponse restResponse = new BytesRestResponse(RestStatus.OK, BytesRestResponse.TEXT_CONTENT_TYPE,
+                data.getBytes(StandardCharsets.UTF_8));
+        if (responseCursor != Cursor.EMPTY) {
+            restResponse.addHeader("Cursor", Cursor.encodeToString(Version.CURRENT, responseCursor));
+        }
+        restResponse.addHeader("Took-nanos", Long.toString(tookNanos));
+        return restResponse;
     }
 
     @Override

sql/server/src/main/java/org/elasticsearch/xpack/sql/session/Cursor.java

@@ -17,6 +17,7 @@ import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.xpack.sql.execution.search.ScrollCursor;
 import org.elasticsearch.xpack.sql.execution.search.extractor.HitExtractors;
+import org.elasticsearch.xpack.sql.plugin.CliFormatterCursor;
 
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
@@ -48,6 +49,7 @@ public interface Cursor extends NamedWriteable {
         entries.addAll(HitExtractors.getNamedWriteables());
         entries.add(new NamedWriteableRegistry.Entry(Cursor.class, EmptyCursor.NAME, in -> EMPTY));
         entries.add(new NamedWriteableRegistry.Entry(Cursor.class, ScrollCursor.NAME, ScrollCursor::new));
+        entries.add(new NamedWriteableRegistry.Entry(Cursor.class, CliFormatterCursor.NAME, CliFormatterCursor::new));
         return entries;
     }