SQL: Add list tables and columns methods to the REST API (elastic/x-pack-elasticsearch#3464)

Adds list tables and list columns methods to the REST API. These methods are needed by JDBC and possibly other clients.

Related elastic/x-pack-elasticsearch#3419

Original commit: elastic/x-pack-elasticsearch@eaa384c7c9

docs/en/sql/endpoints/rest.asciidoc

@@ -184,3 +184,56 @@ or fewer results though. `time_zone` is the time zone to use for date
 functions and date parsing. `time_zone` defaults to `utc` and can take
 any values documented
 http://www.joda.org/joda-time/apidocs/org/joda/time/DateTimeZone.html[here].
+
+
+[[sql-rest-metadata]]
+
+SQL can only work with a subset of Elasticsearch indices. To get the list of
+indices that are available to the user execute:
+
+[source,js]
+--------------------------------------------------
+POST /_xpack/sql/tables
+{
+    "table_pattern": "lib*"
+}
+--------------------------------------------------
+// CONSOLE
+// TEST[setup:library]
+
+Which returns:
+
+[source,js]
+--------------------------------------------------
+{
+    "tables": [ "library" ]
+}
+--------------------------------------------------
+// TESTRESPONSE
+
+
+To get the list of fields that are supported in SQL execute:
+
+[source,js]
+--------------------------------------------------
+POST /_xpack/sql/columns
+{
+    "table_pattern": "library",
+    "column_pattern": ""
+}
+--------------------------------------------------
+// CONSOLE
+// TEST[continued]
+
+[source,js]
+--------------------------------------------------
+{
+    "columns": [
+        {"table": "library", "name": "author",       "type": "text",  "jdbc_type": 12, "display_size": 0},
+        {"table": "library", "name": "name",         "type": "text",  "jdbc_type": 12, "display_size": 0},
+        {"table": "library", "name": "page_count",   "type": "short", "jdbc_type": 5,  "display_size": 6},
+        {"table": "library", "name": "release_date", "type": "date",  "jdbc_type": 93, "display_size": 20}
+    ]
+}
+--------------------------------------------------
+// TESTRESPONSE

plugin/src/main/java/org/elasticsearch/xpack/security/authz/AuthorizationService.java

@@ -67,6 +67,8 @@ import org.elasticsearch.xpack.security.user.XPackSecurityUser;
 import org.elasticsearch.xpack.security.user.XPackUser;
 import org.elasticsearch.xpack.sql.plugin.SqlAction;
 import org.elasticsearch.xpack.sql.plugin.SqlClearCursorAction;
+import org.elasticsearch.xpack.sql.plugin.SqlListColumnsAction;
+import org.elasticsearch.xpack.sql.plugin.SqlListTablesAction;
 import org.elasticsearch.xpack.sql.plugin.SqlTranslateAction;
 
 import java.util.Arrays;
@@ -484,7 +486,9 @@ public class AuthorizationService extends AbstractComponent {
                 action.equals("indices:data/read/search/template") ||
                 action.equals("indices:data/write/reindex") ||
                 action.equals(SqlAction.NAME) ||
-                action.equals(SqlTranslateAction.NAME);
+                action.equals(SqlTranslateAction.NAME) ||
+                action.equals(SqlListTablesAction.NAME) ||
+                action.equals(SqlListColumnsAction.NAME) ;
     }
 
     private static boolean isTranslatedToBulkAction(String action) {

plugin/src/test/java/org/elasticsearch/xpack/sql/SqlActionIT.java

@@ -7,13 +7,24 @@ package org.elasticsearch.xpack.sql;
 
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.support.WriteRequest;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.xpack.sql.plugin.ColumnInfo;
 import org.elasticsearch.xpack.sql.plugin.SqlAction;
+import org.elasticsearch.xpack.sql.plugin.SqlListColumnsAction;
+import org.elasticsearch.xpack.sql.plugin.SqlListColumnsResponse;
+import org.elasticsearch.xpack.sql.plugin.SqlListTablesAction;
+import org.elasticsearch.xpack.sql.plugin.SqlListTablesResponse;
 import org.elasticsearch.xpack.sql.plugin.SqlResponse;
-import org.elasticsearch.xpack.sql.plugin.SqlResponse.ColumnInfo;
 
+import java.io.IOException;
 import java.sql.JDBCType;
+import java.util.List;
+import java.util.stream.Collectors;
 
+import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.hamcrest.Matchers.emptyCollectionOf;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasSize;
 
@@ -35,8 +46,8 @@ public class SqlActionIT extends AbstractSqlIntegTestCase {
         assertThat(response.columns(), hasSize(2));
         int dataIndex = dataBeforeCount ? 0 : 1;
         int countIndex = dataBeforeCount ? 1 : 0;
-        assertEquals(new ColumnInfo("data", "text", JDBCType.VARCHAR, 0), response.columns().get(dataIndex));
+        assertEquals(new ColumnInfo("", "data", "text", JDBCType.VARCHAR, 0), response.columns().get(dataIndex));
-        assertEquals(new ColumnInfo("count", "long", JDBCType.BIGINT, 20), response.columns().get(countIndex));
+        assertEquals(new ColumnInfo("", "count", "long", JDBCType.BIGINT, 20), response.columns().get(countIndex));
 
         assertThat(response.rows(), hasSize(2));
         assertEquals("bar", response.rows().get(0).get(dataIndex));
@@ -44,5 +55,76 @@ public class SqlActionIT extends AbstractSqlIntegTestCase {
         assertEquals("baz", response.rows().get(1).get(dataIndex));
         assertEquals(43L, response.rows().get(1).get(countIndex));
     }
+
+    public void testSqlListTablesAction() throws Exception {
+
+        createCompatibleIndex("foo");
+        createCompatibleIndex("bar");
+        createCompatibleIndex("baz");
+        createIncompatibleIndex("broken");
+
+        SqlListTablesResponse response = client().prepareExecute(SqlListTablesAction.INSTANCE)
+                .pattern("").get();
+        List<String> tables = removeInternal(response.getTables());
+        assertThat(tables, hasSize(3));
+        assertThat(tables, containsInAnyOrder("foo", "bar", "baz"));
+
+
+        response = client().prepareExecute(SqlListTablesAction.INSTANCE).pattern("b*").get();
+        tables = removeInternal(response.getTables());
+        assertThat(tables, hasSize(2));
+        assertThat(tables, containsInAnyOrder("bar", "baz"));
+
+        response = client().prepareExecute(SqlListTablesAction.INSTANCE).pattern("not_found").get();
+        tables = removeInternal(response.getTables());
+        assertThat(tables, emptyCollectionOf(String.class));
+
+        response = client().prepareExecute(SqlListTablesAction.INSTANCE).pattern("broken").get();
+        tables = removeInternal(response.getTables());
+        assertThat(tables, emptyCollectionOf(String.class));
+    }
+
+
+    public void testSqlListColumnsAction() throws Exception {
+
+        createCompatibleIndex("bar");
+        createCompatibleIndex("baz");
+        createIncompatibleIndex("broken");
+
+        SqlListColumnsResponse response = client().prepareExecute(SqlListColumnsAction.INSTANCE)
+                .indexPattern("bar").columnPattern("").get();
+        List<ColumnInfo> columns = response.getColumns();
+        assertThat(columns, hasSize(2));
+        assertThat(columns, containsInAnyOrder(
+                new ColumnInfo("bar", "str_field", "text", JDBCType.VARCHAR, 0),
+                new ColumnInfo("bar", "int_field", "integer", JDBCType.INTEGER, 11)
+        ));
+    }
+
+    private void createCompatibleIndex(String name) throws IOException {
+        XContentBuilder mapping = jsonBuilder().startObject();
+        {
+            mapping.startObject("properties");
+            {
+                mapping.startObject("str_field").field("type", "text").endObject();
+                mapping.startObject("int_field").field("type", "integer").endObject();
+            }
+            mapping.endObject();
+        }
+        mapping.endObject();
+
+        assertAcked(client().admin().indices().prepareCreate(name).addMapping("doc", mapping).get());
+    }
+
+    private void createIncompatibleIndex(String name) throws IOException {
+        assertAcked(client().admin().indices().prepareCreate(name).get());
+    }
+
+    /**
+     * Removes list of internal indices to make tests consistent between secure and unsecure environments
+     */
+    private static List<String> removeInternal(List<String> list) {
+        return list.stream().filter(s -> s.startsWith(".") == false).collect(Collectors.toList());
+    }
 }
 

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

@@ -0,0 +1,16 @@
+{
+    "xpack.sql.columns": {
+      "documentation": "Returns a list of columns supported by SQL for the current user",
+      "methods": [ "POST" ],
+      "url": {
+        "path": "/_xpack/sql/columns",
+        "paths": [ "/_xpack/sql/columns" ],
+        "parts": {},
+        "params": {}
+      },
+      "body": {
+        "description" : "Specify the table pattern in the `table_pattern` and the column patter in `column_pattern` element.",
+        "required" : true
+      }
+    }
+  }

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

@@ -0,0 +1,16 @@
+{
+    "xpack.sql.tables": {
+      "documentation": "Returns a list of tables supported by SQL for the current user",
+      "methods": [ "POST" ],
+      "url": {
+        "path": "/_xpack/sql/tables",
+        "paths": [ "/_xpack/sql/tables" ],
+        "parts": {},
+        "params": {}
+      },
+      "body": {
+        "description" : "Specify the table pattern in the `table_pattern` element.",
+        "required" : true
+      }
+    }
+  }

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

@@ -118,3 +118,24 @@ setup:
 
   - match: { indices.test.total.search.open_contexts: 0 }
 
+---
+"Check list of tables":
+  - do:
+      xpack.sql.tables:
+        body:
+          table_pattern: "t*"
+
+  - length: { tables: 1 }
+  - match: { tables.0: 'test' }
+
+---
+"Check list of columns":
+  - do:
+      xpack.sql.columns:
+        body:
+          table_pattern: "t*"
+          column_pattern: ""
+
+  - length: { columns: 2 }
+  - match: { columns.0.name: 'int' }
+  - match: { columns.1.name: 'str' }

sql/cli/src/test/java/org/elasticsearch/xpack/sql/cli/command/ServerQueryCliCommandTests.java

@@ -8,6 +8,7 @@ package org.elasticsearch.xpack.sql.cli.command;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xpack.sql.cli.TestTerminal;
 import org.elasticsearch.xpack.sql.client.HttpClient;
+import org.elasticsearch.xpack.sql.plugin.ColumnInfo;
 import org.elasticsearch.xpack.sql.plugin.SqlResponse;
 
 import java.sql.JDBCType;
@@ -105,9 +106,9 @@ public class ServerQueryCliCommandTests extends ESTestCase {
 
     private SqlResponse fakeResponse(String cursor, boolean includeColumns, String val) {
         List<List<Object>> rows;
-        List<SqlResponse.ColumnInfo> columns;
+        List<ColumnInfo> columns;
         if (includeColumns) {
-            columns = Collections.singletonList(new SqlResponse.ColumnInfo("field", "string", JDBCType.VARCHAR, 0));
+            columns = Collections.singletonList(new ColumnInfo("", "field", "string", JDBCType.VARCHAR, 0));
         } else {
             columns = null;
         }

sql/rest-proto/src/main/java/org/elasticsearch/xpack/sql/plugin/ColumnInfo.java

@@ -0,0 +1,167 @@
+/*
+ * 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.common.ParseField;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.xcontent.ConstructingObjectParser;
+import org.elasticsearch.common.xcontent.ToXContentObject;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentParser;
+
+import java.io.IOException;
+import java.sql.JDBCType;
+import java.util.Objects;
+
+import static org.elasticsearch.common.xcontent.ConstructingObjectParser.constructorArg;
+import static org.elasticsearch.common.xcontent.ConstructingObjectParser.optionalConstructorArg;
+
+/**
+ * Information about a column.
+ */
+public final class ColumnInfo implements Writeable, ToXContentObject {
+    public static final String JDBC_ENABLED_PARAM = "jdbc_enabled";
+    public static final int UNKNOWN_DISPLAY_SIZE = -1;
+
+    private static final ConstructingObjectParser<ColumnInfo, Void> PARSER =
+            new ConstructingObjectParser<>("column_info", true, objects ->
+                    new ColumnInfo(
+                            objects[0] == null ? "" : (String) objects[0],
+                            (String) objects[1],
+                            (String) objects[2],
+                            objects[3] == null ? null : JDBCType.valueOf((int) objects[3]),
+                            objects[4] == null ? UNKNOWN_DISPLAY_SIZE : (int) objects[4]));
+
+    private static final ParseField TABLE = new ParseField("table");
+    private static final ParseField NAME = new ParseField("name");
+    private static final ParseField ES_TYPE = new ParseField("type");
+    private static final ParseField JDBC_TYPE = new ParseField("jdbc_type");
+    private static final ParseField DISPLAY_SIZE = new ParseField("display_size");
+
+    static {
+        PARSER.declareString(optionalConstructorArg(), TABLE);
+        PARSER.declareString(constructorArg(), NAME);
+        PARSER.declareString(constructorArg(), ES_TYPE);
+        PARSER.declareInt(optionalConstructorArg(), JDBC_TYPE);
+        PARSER.declareInt(optionalConstructorArg(), DISPLAY_SIZE);
+    }
+
+    private final String table;
+    private final String name;
+    private final String esType;
+    private final JDBCType jdbcType;
+    private final int displaySize;
+
+    public ColumnInfo(String table, String name, String esType, JDBCType jdbcType, int displaySize) {
+        this.table = table;
+        this.name = name;
+        this.esType = esType;
+        this.jdbcType = jdbcType;
+        this.displaySize = displaySize;
+    }
+
+    ColumnInfo(StreamInput in) throws IOException {
+        table = in.readString();
+        name = in.readString();
+        esType = in.readString();
+        jdbcType = JDBCType.valueOf(in.readVInt());
+        displaySize = in.readVInt();
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeString(table);
+        out.writeString(name);
+        out.writeString(esType);
+        out.writeVInt(jdbcType.getVendorTypeNumber());
+        out.writeVInt(displaySize);
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        return toXContent(builder, params.paramAsBoolean(JDBC_ENABLED_PARAM, true));
+    }
+
+    public XContentBuilder toXContent(XContentBuilder builder, boolean isJdbcAllowed) throws IOException {
+        builder.startObject();
+        if (Strings.hasText(table)) {
+            builder.field("table", table);
+        }
+        builder.field("name", name);
+        builder.field("type", esType);
+        if (isJdbcAllowed && jdbcType != null) {
+            builder.field("jdbc_type", jdbcType.getVendorTypeNumber());
+            builder.field("display_size", displaySize);
+        }
+        return builder.endObject();
+    }
+
+
+    public static ColumnInfo fromXContent(XContentParser parser) {
+        return PARSER.apply(parser, null);
+    }
+
+    /**
+     * Name of the table.
+     */
+    public String table() {
+        return table;
+    }
+
+    /**
+     * Name of the column.
+     */
+    public String name() {
+        return name;
+    }
+
+    /**
+     * The type of the column in Elasticsearch.
+     */
+    public String esType() {
+        return esType;
+    }
+
+    /**
+     * The type of the column as it would be returned by a JDBC driver.
+     */
+    public JDBCType jdbcType() {
+        return jdbcType;
+    }
+
+    /**
+     * Used by JDBC
+     */
+    public int displaySize() {
+        return displaySize;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (obj == null || obj.getClass() != getClass()) {
+            return false;
+        }
+        ColumnInfo other = (ColumnInfo) obj;
+        return table.equals(other.table)
+                && name.equals(other.name)
+                && esType.equals(other.esType)
+                && jdbcType.equals(other.jdbcType)
+                && displaySize == other.displaySize;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(table, name, esType, jdbcType, displaySize);
+    }
+
+    @Override
+    public String toString() {
+        return Strings.toString(this);
+    }
+}

sql/rest-proto/src/main/java/org/elasticsearch/xpack/sql/plugin/SqlListColumnsAction.java

@@ -0,0 +1,30 @@
+/*
+ * 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.Action;
+import org.elasticsearch.client.ElasticsearchClient;
+
+public class SqlListColumnsAction extends Action<SqlListColumnsRequest, SqlListColumnsResponse, SqlListColumnsRequestBuilder> {
+
+    public static final SqlListColumnsAction INSTANCE = new SqlListColumnsAction();
+    public static final String NAME = "indices:admin/sql/columns";
+    public static final String REST_ENDPOINT = "/_xpack/sql/columns";
+
+    private SqlListColumnsAction() {
+        super(NAME);
+    }
+
+    @Override
+    public SqlListColumnsRequestBuilder newRequestBuilder(ElasticsearchClient client) {
+        return new SqlListColumnsRequestBuilder(client, this);
+    }
+
+    @Override
+    public SqlListColumnsResponse newResponse() {
+        return new SqlListColumnsResponse();
+    }
+}

sql/rest-proto/src/main/java/org/elasticsearch/xpack/sql/plugin/SqlListColumnsRequest.java

@@ -0,0 +1,144 @@
+/*
+ * 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.ActionRequest;
+import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.action.CompositeIndicesRequest;
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.xcontent.ConstructingObjectParser;
+import org.elasticsearch.common.xcontent.ToXContentObject;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentParser;
+
+import java.io.IOException;
+import java.util.Objects;
+
+import static org.elasticsearch.action.ValidateActions.addValidationError;
+import static org.elasticsearch.common.xcontent.ConstructingObjectParser.constructorArg;
+
+/**
+ * Request to get a list of SQL-supported columns of an index
+ * <p>
+ * It needs to be CompositeIndicesRequest because we resolve wildcards a non-standard SQL
+ * manner
+ */
+public class SqlListColumnsRequest extends ActionRequest implements ToXContentObject, CompositeIndicesRequest {
+    @SuppressWarnings("unchecked")
+    public static final ConstructingObjectParser<SqlListColumnsRequest, Void> PARSER =
+            new ConstructingObjectParser<>("sql_list_tables", true, objects -> new SqlListColumnsRequest(
+                    (String) objects[0],
+                    (String) objects[1]
+            ));
+
+    public static final ParseField TABLE_PATTERN = new ParseField("table_pattern");
+    public static final ParseField COLUMN_PATTERN = new ParseField("column_pattern");
+
+    static {
+        PARSER.declareString(constructorArg(), TABLE_PATTERN);
+        PARSER.declareString(constructorArg(), COLUMN_PATTERN);
+    }
+
+    private String tablePattern;
+    private String columnPattern;
+
+
+    public SqlListColumnsRequest() {
+    }
+
+    public SqlListColumnsRequest(String tablePattern, String columnPattern) {
+        this.tablePattern = tablePattern;
+        this.columnPattern = columnPattern;
+    }
+
+
+    public SqlListColumnsRequest(StreamInput in) throws IOException {
+        super(in);
+        this.tablePattern = in.readString();
+        this.columnPattern = in.readString();
+    }
+
+    @Override
+    public ActionRequestValidationException validate() {
+        ActionRequestValidationException validationException = null;
+        if (tablePattern == null) {
+            validationException = addValidationError("[index_pattern] is required", validationException);
+        }
+        if (columnPattern == null) {
+            validationException = addValidationError("[column_pattern] is required", validationException);
+        }
+        return validationException;
+    }
+
+    /**
+     * The index pattern for the results
+     */
+    public String getTablePattern() {
+        return tablePattern;
+    }
+
+    public void setTablePattern(String tablePattern) {
+        this.tablePattern = tablePattern;
+    }
+
+    /**
+     * The column pattern for the results
+     */
+    public String getColumnPattern() {
+        return columnPattern;
+    }
+
+    public void setColumnPattern(String columnPattern) {
+        this.columnPattern = columnPattern;
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeString(tablePattern);
+        out.writeString(columnPattern);
+    }
+
+    @Override
+    public void readFrom(StreamInput in) throws IOException {
+        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        SqlListColumnsRequest that = (SqlListColumnsRequest) o;
+        return Objects.equals(tablePattern, that.tablePattern) &&
+                Objects.equals(columnPattern, that.columnPattern);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(tablePattern, columnPattern);
+    }
+
+    @Override
+    public String getDescription() {
+        return "SQL List Columns[" + getTablePattern() + ", " + getColumnPattern() + "]";
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject();
+        {
+            builder.field("table_pattern", tablePattern);
+            builder.field("column_pattern", columnPattern);
+        }
+        return builder.endObject();
+    }
+
+    public static SqlListColumnsRequest fromXContent(XContentParser parser) {
+        return PARSER.apply(parser, null);
+    }
+}

sql/rest-proto/src/main/java/org/elasticsearch/xpack/sql/plugin/SqlListColumnsRequestBuilder.java

@@ -0,0 +1,32 @@
+/*
+ * 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.ActionRequestBuilder;
+import org.elasticsearch.client.ElasticsearchClient;
+
+public class SqlListColumnsRequestBuilder extends
+        ActionRequestBuilder<SqlListColumnsRequest, SqlListColumnsResponse, SqlListColumnsRequestBuilder> {
+
+    public SqlListColumnsRequestBuilder(ElasticsearchClient client, SqlListColumnsAction action, 
+                                        String indexPattern, String columnPattern) {
+        super(client, action, new SqlListColumnsRequest(indexPattern, columnPattern));
+    }
+
+    public SqlListColumnsRequestBuilder(ElasticsearchClient client, SqlListColumnsAction action) {
+        super(client, action, new SqlListColumnsRequest());
+    }
+
+    public SqlListColumnsRequestBuilder indexPattern(String indexPattern) {
+        request.setTablePattern(indexPattern);
+        return this;
+    }
+
+    public SqlListColumnsRequestBuilder columnPattern(String columnPattern) {
+        request.setColumnPattern(columnPattern);
+        return this;
+    }
+}

sql/rest-proto/src/main/java/org/elasticsearch/xpack/sql/plugin/SqlListColumnsResponse.java

@@ -0,0 +1,107 @@
+/*
+ * 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.ActionResponse;
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.xcontent.ConstructingObjectParser;
+import org.elasticsearch.common.xcontent.ToXContentObject;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentParser;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Objects;
+
+import static org.elasticsearch.common.xcontent.ConstructingObjectParser.optionalConstructorArg;
+import static org.elasticsearch.xpack.sql.plugin.ColumnInfo.JDBC_ENABLED_PARAM;
+
+/**
+ * Response to perform an sql query
+ */
+public class SqlListColumnsResponse extends ActionResponse implements ToXContentObject {
+    @SuppressWarnings("unchecked")
+    public static final ConstructingObjectParser<SqlListColumnsResponse, Void> PARSER = new ConstructingObjectParser<>("sql", true,
+            objects -> new SqlListColumnsResponse((List<ColumnInfo>) objects[0]));
+
+    public static final ParseField COLUMNS = new ParseField("columns");
+
+    static {
+        PARSER.declareObjectArray(optionalConstructorArg(), (p, c) -> ColumnInfo.fromXContent(p), COLUMNS);
+    }
+
+    private List<ColumnInfo> columns;
+
+    public SqlListColumnsResponse() {
+    }
+
+    public SqlListColumnsResponse(List<ColumnInfo> columns) {
+        this.columns = columns;
+    }
+
+    /**
+     * The key that must be sent back to SQL to access the next page of
+     * results. If equal to "" then there is no next page.
+     */
+    public List<ColumnInfo> getColumns() {
+        return columns;
+    }
+
+    @Override
+    public void readFrom(StreamInput in) throws IOException {
+        super.readFrom(in);
+        columns = in.readList(ColumnInfo::new);
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeList(columns);
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        boolean isJdbcAllowed = params.paramAsBoolean(JDBC_ENABLED_PARAM, true);
+
+        builder.startObject();
+        {
+            builder.startArray("columns");
+            {
+                for (ColumnInfo column : columns) {
+                    column.toXContent(builder, isJdbcAllowed);
+                }
+            }
+            builder.endArray();
+        }
+        return builder.endObject();
+    }
+
+    public static SqlListColumnsResponse fromXContent(XContentParser parser) {
+        return PARSER.apply(parser, null);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        SqlListColumnsResponse that = (SqlListColumnsResponse) o;
+        return Objects.equals(columns, that.columns);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(columns);
+    }
+
+    @Override
+    public String toString() {
+        return Strings.toString(this);
+    }
+
+}

sql/rest-proto/src/main/java/org/elasticsearch/xpack/sql/plugin/SqlListTablesAction.java

@@ -0,0 +1,30 @@
+/*
+ * 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.Action;
+import org.elasticsearch.client.ElasticsearchClient;
+
+public class SqlListTablesAction extends Action<SqlListTablesRequest, SqlListTablesResponse, SqlListTablesRequestBuilder> {
+
+    public static final SqlListTablesAction INSTANCE = new SqlListTablesAction();
+    public static final String NAME = "indices:admin/sql/tables";
+    public static final String REST_ENDPOINT = "/_xpack/sql/tables";
+
+    private SqlListTablesAction() {
+        super(NAME);
+    }
+
+    @Override
+    public SqlListTablesRequestBuilder newRequestBuilder(ElasticsearchClient client) {
+        return new SqlListTablesRequestBuilder(client, this);
+    }
+
+    @Override
+    public SqlListTablesResponse newResponse() {
+        return new SqlListTablesResponse();
+    }
+}

sql/rest-proto/src/main/java/org/elasticsearch/xpack/sql/plugin/SqlListTablesRequest.java

@@ -0,0 +1,119 @@
+/*
+ * 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.ActionRequest;
+import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.action.CompositeIndicesRequest;
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.xcontent.ConstructingObjectParser;
+import org.elasticsearch.common.xcontent.ToXContentObject;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentParser;
+
+import java.io.IOException;
+import java.util.Objects;
+
+import static org.elasticsearch.action.ValidateActions.addValidationError;
+import static org.elasticsearch.common.xcontent.ConstructingObjectParser.constructorArg;
+
+/**
+ * Request to get a list of SQL-supported indices
+ * <p>
+ * It needs to be CompositeIndicesRequest because we resolve wildcards a non-standard SQL
+ * manner
+ */
+public class SqlListTablesRequest extends ActionRequest implements ToXContentObject, CompositeIndicesRequest {
+    @SuppressWarnings("unchecked")
+    public static final ConstructingObjectParser<SqlListTablesRequest, Void> PARSER
+            = new ConstructingObjectParser<>("sql_list_tables", true, objects -> new SqlListTablesRequest((String) objects[0]));
+
+    public static final ParseField TABLE_PATTERN = new ParseField("table_pattern");
+
+    static {
+        PARSER.declareString(constructorArg(), TABLE_PATTERN);
+    }
+
+    private String pattern;
+
+    public SqlListTablesRequest() {
+    }
+
+    public SqlListTablesRequest(String pattern) {
+        this.pattern = pattern;
+    }
+
+
+    public SqlListTablesRequest(StreamInput in) throws IOException {
+        super(in);
+        this.pattern = in.readString();
+    }
+
+    @Override
+    public ActionRequestValidationException validate() {
+        ActionRequestValidationException validationException = null;
+        if (pattern == null) {
+            validationException = addValidationError("[pattern] is required", validationException);
+        }
+        return validationException;
+    }
+
+    /**
+     * The pattern for the results
+     */
+    public String getPattern() {
+        return pattern;
+    }
+
+    public void setPattern(String pattern) {
+        this.pattern = pattern;
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeString(pattern);
+    }
+
+    @Override
+    public void readFrom(StreamInput in) throws IOException {
+
+        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        SqlListTablesRequest that = (SqlListTablesRequest) o;
+        return Objects.equals(pattern, that.pattern);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(pattern);
+    }
+
+    @Override
+    public String getDescription() {
+        return "SQL List Tables[" + getPattern() + "]";
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject();
+        {
+            builder.field("table_pattern", pattern);
+        }
+        return builder.endObject();
+    }
+
+    public static SqlListTablesRequest fromXContent(XContentParser parser) {
+        return PARSER.apply(parser, null);
+    }
+}

sql/rest-proto/src/main/java/org/elasticsearch/xpack/sql/plugin/SqlListTablesRequestBuilder.java

@@ -0,0 +1,26 @@
+/*
+ * 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.ActionRequestBuilder;
+import org.elasticsearch.client.ElasticsearchClient;
+
+public class SqlListTablesRequestBuilder extends
+        ActionRequestBuilder<SqlListTablesRequest, SqlListTablesResponse, SqlListTablesRequestBuilder> {
+
+    public SqlListTablesRequestBuilder(ElasticsearchClient client, SqlListTablesAction action, String pattern) {
+        super(client, action, new SqlListTablesRequest(pattern));
+    }
+
+    public SqlListTablesRequestBuilder(ElasticsearchClient client, SqlListTablesAction action) {
+        super(client, action, new SqlListTablesRequest());
+    }
+
+    public SqlListTablesRequestBuilder pattern(String pattern) {
+        request.setPattern(pattern);
+        return this;
+    }
+}

sql/rest-proto/src/main/java/org/elasticsearch/xpack/sql/plugin/SqlListTablesResponse.java

@@ -0,0 +1,105 @@
+/*
+ * 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.ActionResponse;
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.xcontent.ConstructingObjectParser;
+import org.elasticsearch.common.xcontent.ToXContentObject;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentParser;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Objects;
+
+import static org.elasticsearch.common.xcontent.ConstructingObjectParser.optionalConstructorArg;
+
+/**
+ * Response to perform an sql query
+ */
+public class SqlListTablesResponse extends ActionResponse implements ToXContentObject {
+    @SuppressWarnings("unchecked")
+    public static final ConstructingObjectParser<SqlListTablesResponse, Void> PARSER =
+            new ConstructingObjectParser<>("sql_list_tables", true,
+                    objects -> new SqlListTablesResponse((List<String>) objects[0]));
+
+    public static final ParseField TABLES = new ParseField("tables");
+
+    static {
+        PARSER.declareStringArray(optionalConstructorArg(), TABLES);
+    }
+
+    private List<String> tables;
+
+    public SqlListTablesResponse() {
+    }
+
+    public SqlListTablesResponse(List<String> tables) {
+        this.tables = tables;
+    }
+
+    /**
+     * The key that must be sent back to SQL to access the next page of
+     * results. If equal to "" then there is no next page.
+     */
+    public List<String> getTables() {
+        return tables;
+    }
+
+    @Override
+    public void readFrom(StreamInput in) throws IOException {
+        super.readFrom(in);
+        tables = in.readList(StreamInput::readString);
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeStringList(tables);
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject();
+        {
+            builder.startArray("tables");
+            {
+                for (String table : tables) {
+                    builder.value(table);
+                }
+            }
+            builder.endArray();
+        }
+        return builder.endObject();
+    }
+
+    public static SqlListTablesResponse fromXContent(XContentParser parser) {
+        return PARSER.apply(parser, null);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        SqlListTablesResponse that = (SqlListTablesResponse) o;
+        return Objects.equals(tables, that.tables);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(tables);
+    }
+
+    @Override
+    public String toString() {
+        return Strings.toString(this);
+    }
+
+}

sql/rest-proto/src/main/java/org/elasticsearch/xpack/sql/plugin/SqlResponse.java

@@ -28,13 +28,12 @@ import static java.util.Collections.unmodifiableList;
 import static org.elasticsearch.common.xcontent.ConstructingObjectParser.constructorArg;
 import static org.elasticsearch.common.xcontent.ConstructingObjectParser.optionalConstructorArg;
 import static org.elasticsearch.common.xcontent.XContentParserUtils.parseStoredFieldsValue;
+import static org.elasticsearch.xpack.sql.plugin.ColumnInfo.JDBC_ENABLED_PARAM;
 
 /**
  * Response to perform an sql query
  */
 public class SqlResponse extends ActionResponse implements ToXContentObject {
-    public static final String JDBC_ENABLED_PARAM = "jdbc_enabled";
-    public static final int UNKNOWN_DISPLAY_SIZE = -1;
 
     @SuppressWarnings("unchecked")
     public static final ConstructingObjectParser<SqlResponse, Void> PARSER = new ConstructingObjectParser<>("sql", true,
@@ -236,128 +235,4 @@ public class SqlResponse extends ActionResponse implements ToXContentObject {
     public String toString() {
         return Strings.toString(this);
     }
-
-
-    private static final ConstructingObjectParser<ColumnInfo, Void> COLUMN_INFO_PARSER =
-            new ConstructingObjectParser<>("sql", true, objects ->
-                    new ColumnInfo(
-                            (String) objects[0],
-                            (String) objects[1],
-                            objects[2] == null ? null : JDBCType.valueOf((int) objects[2]),
-                            objects[3] == null ? UNKNOWN_DISPLAY_SIZE : (int) objects[3]));
-
-    private static final ParseField NAME = new ParseField("name");
-    private static final ParseField ES_TYPE = new ParseField("type");
-    private static final ParseField JDBC_TYPE = new ParseField("jdbc_type");
-    private static final ParseField DISPLAY_SIZE = new ParseField("display_size");
-
-    static {
-        COLUMN_INFO_PARSER.declareString(constructorArg(), NAME);
-        COLUMN_INFO_PARSER.declareString(constructorArg(), ES_TYPE);
-        COLUMN_INFO_PARSER.declareInt(optionalConstructorArg(), JDBC_TYPE);
-        COLUMN_INFO_PARSER.declareInt(optionalConstructorArg(), DISPLAY_SIZE);
-    }
-
-    /**
-     * Information about a column.
-     */
-    public static final class ColumnInfo implements Writeable, ToXContentObject {
-        private final String name;
-        private final String esType;
-        private final JDBCType jdbcType;
-        private final int displaySize;
-
-        public ColumnInfo(String name, String esType, JDBCType jdbcType, int displaySize) {
-            this.name = name;
-            this.esType = esType;
-            this.jdbcType = jdbcType;
-            this.displaySize = displaySize;
-        }
-
-        ColumnInfo(StreamInput in) throws IOException {
-            name = in.readString();
-            esType = in.readString();
-            jdbcType = JDBCType.valueOf(in.readVInt());
-            displaySize = in.readVInt();
-        }
-
-        @Override
-        public void writeTo(StreamOutput out) throws IOException {
-            out.writeString(name);
-            out.writeString(esType);
-            out.writeVInt(jdbcType.getVendorTypeNumber());
-            out.writeVInt(displaySize);
-        }
-
-        @Override
-        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-            return toXContent(builder, params.paramAsBoolean(JDBC_ENABLED_PARAM, true));
-        }
-
-        public XContentBuilder toXContent(XContentBuilder builder, boolean isJdbcAllowed) throws IOException {
-            builder.startObject();
-            builder.field("name", name);
-            builder.field("type", esType);
-            if (isJdbcAllowed && jdbcType != null) {
-                builder.field("jdbc_type", jdbcType.getVendorTypeNumber());
-                builder.field("display_size", displaySize);
-            }
-            return builder.endObject();
-        }
-
-
-        public static ColumnInfo fromXContent(XContentParser parser) {
-            return COLUMN_INFO_PARSER.apply(parser, null);
-        }
-
-        /**
-         * Name of the column.
-         */
-        public String name() {
-            return name;
-        }
-
-        /**
-         * The type of the column in Elasticsearch.
-         */
-        public String esType() {
-            return esType;
-        }
-
-        /**
-         * The type of the column as it would be returned by a JDBC driver.
-         */
-        public JDBCType jdbcType() {
-            return jdbcType;
-        }
-
-        /**
-         * Used by JDBC
-         */
-        public int displaySize() {
-            return displaySize;
-        }
-
-        @Override
-        public boolean equals(Object obj) {
-            if (obj == null || obj.getClass() != getClass()) {
-                return false;
-            }
-            ColumnInfo other = (ColumnInfo) obj;
-            return name.equals(other.name)
-                    && esType.equals(other.esType)
-                    && jdbcType.equals(other.jdbcType)
-                    && displaySize == other.displaySize;
-        }
-
-        @Override
-        public int hashCode() {
-            return Objects.hash(name, esType, jdbcType, displaySize);
-        }
-
-        @Override
-        public String toString() {
-            return Strings.toString(this);
-        }
-    }
 }

sql/rest-proto/src/test/java/org/elasticsearch/xpack/sql/plugin/SqlListColumnsRequestTests.java

@@ -0,0 +1,29 @@
+/*
+ * 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.common.io.stream.Writeable;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.test.AbstractSerializingTestCase;
+
+import java.io.IOException;
+
+public class SqlListColumnsRequestTests extends AbstractSerializingTestCase<SqlListColumnsRequest> {
+    @Override
+    protected SqlListColumnsRequest createTestInstance() {
+        return new SqlListColumnsRequest(randomAlphaOfLength(10), randomAlphaOfLength(10));
+    }
+
+    @Override
+    protected Writeable.Reader<SqlListColumnsRequest> instanceReader() {
+        return SqlListColumnsRequest::new;
+    }
+
+    @Override
+    protected SqlListColumnsRequest doParseInstance(XContentParser parser) throws IOException {
+        return SqlListColumnsRequest.fromXContent(parser);
+    }
+}

sql/rest-proto/src/test/java/org/elasticsearch/xpack/sql/plugin/SqlListColumnsResponseTests.java

@@ -0,0 +1,82 @@
+/*
+ * 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.common.xcontent.ToXContent;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentFactory;
+import org.elasticsearch.common.xcontent.XContentHelper;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.test.AbstractStreamableXContentTestCase;
+
+import java.io.IOException;
+import java.sql.JDBCType;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import static org.elasticsearch.xpack.sql.plugin.ColumnInfo.JDBC_ENABLED_PARAM;
+import static org.hamcrest.Matchers.hasSize;
+
+public class SqlListColumnsResponseTests extends AbstractStreamableXContentTestCase<SqlListColumnsResponse> {
+
+    @Override
+    protected SqlListColumnsResponse createTestInstance() {
+        int columnCount = between(1, 10);
+
+        List<ColumnInfo> columns = new ArrayList<>(columnCount);
+        for (int i = 0; i < columnCount; i++) {
+            columns.add(new ColumnInfo(randomAlphaOfLength(10), randomAlphaOfLength(10), randomAlphaOfLength(10),
+                    randomFrom(JDBCType.values()), randomInt(25)));
+        }
+        return new SqlListColumnsResponse(columns);
+    }
+
+    @Override
+    protected SqlListColumnsResponse createBlankInstance() {
+        return new SqlListColumnsResponse();
+    }
+
+    public void testToXContent() throws IOException {
+        SqlListColumnsResponse testInstance = createTestInstance();
+
+        boolean jdbcEnabled = randomBoolean();
+        ToXContent.Params params =
+                new ToXContent.MapParams(Collections.singletonMap(JDBC_ENABLED_PARAM, Boolean.toString(jdbcEnabled)));
+
+        XContentBuilder builder = testInstance.toXContent(XContentFactory.jsonBuilder(), params);
+        Map<String, Object> rootMap = XContentHelper.convertToMap(builder.bytes(), false, builder.contentType()).v2();
+
+        logger.info(builder.string());
+
+        if (testInstance.getColumns() != null) {
+            List<?> columns = (List<?>) rootMap.get("columns");
+            assertThat(columns, hasSize(testInstance.getColumns().size()));
+            for (int i = 0; i < columns.size(); i++) {
+                Map<?, ?> columnMap = (Map<?, ?>) columns.get(i);
+                ColumnInfo columnInfo = testInstance.getColumns().get(i);
+                assertEquals(columnInfo.table(), columnMap.get("table"));
+                assertEquals(columnInfo.name(), columnMap.get("name"));
+                assertEquals(columnInfo.esType(), columnMap.get("type"));
+                if (jdbcEnabled) {
+                    assertEquals(columnInfo.displaySize(), columnMap.get("display_size"));
+                    assertEquals(columnInfo.jdbcType().getVendorTypeNumber(), columnMap.get("jdbc_type"));
+                } else {
+                    assertNull(columnMap.get("display_size"));
+                    assertNull(columnMap.get("jdbc_type"));
+                }
+            }
+        } else {
+            assertNull(rootMap.get("columns"));
+        }
+    }
+
+    @Override
+    protected SqlListColumnsResponse doParseInstance(XContentParser parser) {
+        return SqlListColumnsResponse.fromXContent(parser);
+    }
+}

sql/rest-proto/src/test/java/org/elasticsearch/xpack/sql/plugin/SqlListTablesRequestTests.java

@@ -0,0 +1,30 @@
+/*
+ * 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.common.io.stream.Writeable;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.test.AbstractSerializingTestCase;
+import org.elasticsearch.test.AbstractWireSerializingTestCase;
+
+import java.io.IOException;
+
+public class SqlListTablesRequestTests extends AbstractSerializingTestCase<SqlListTablesRequest> {
+    @Override
+    protected SqlListTablesRequest createTestInstance() {
+        return new SqlListTablesRequest(randomAlphaOfLength(10));
+    }
+
+    @Override
+    protected Writeable.Reader<SqlListTablesRequest> instanceReader() {
+        return SqlListTablesRequest::new;
+    }
+
+    @Override
+    protected SqlListTablesRequest doParseInstance(XContentParser parser) throws IOException {
+        return SqlListTablesRequest.fromXContent(parser);
+    }
+}

sql/rest-proto/src/test/java/org/elasticsearch/xpack/sql/plugin/SqlListTablesResponseTests.java

@@ -0,0 +1,36 @@
+/*
+ * 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.common.xcontent.XContentParser;
+import org.elasticsearch.test.AbstractStreamableXContentTestCase;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class SqlListTablesResponseTests extends AbstractStreamableXContentTestCase<SqlListTablesResponse> {
+
+    @Override
+    protected SqlListTablesResponse createTestInstance() {
+        int tableCount = between(1, 10);
+
+        List<String> tables = new ArrayList<>(tableCount);
+        for (int i = 0; i < tableCount; i++) {
+            tables.add(randomAlphaOfLength(10));
+        }
+        return new SqlListTablesResponse(tables);
+    }
+
+    @Override
+    protected SqlListTablesResponse createBlankInstance() {
+        return new SqlListTablesResponse();
+    }
+
+    @Override
+    protected SqlListTablesResponse doParseInstance(XContentParser parser) {
+        return SqlListTablesResponse.fromXContent(parser);
+    }
+}

sql/rest-proto/src/test/java/org/elasticsearch/xpack/sql/plugin/SqlResponseTests.java

@@ -12,7 +12,6 @@ import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.test.AbstractStreamableXContentTestCase;
 import org.elasticsearch.test.ESTestCase;
-import org.elasticsearch.xpack.sql.plugin.SqlResponse.ColumnInfo;
 
 import java.io.IOException;
 import java.sql.JDBCType;
@@ -23,6 +22,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.function.Supplier;
 
+import static org.elasticsearch.xpack.sql.plugin.ColumnInfo.JDBC_ENABLED_PARAM;
 import static org.hamcrest.Matchers.hasSize;
 
 public class SqlResponseTests extends AbstractStreamableXContentTestCase<SqlResponse> {
@@ -43,7 +43,8 @@ public class SqlResponseTests extends AbstractStreamableXContentTestCase<SqlResp
         if (randomBoolean()) {
             columns = new ArrayList<>(columnCount);
             for (int i = 0; i < columnCount; i++) {
-                columns.add(new ColumnInfo(randomAlphaOfLength(10), randomAlphaOfLength(10), randomFrom(JDBCType.values()), randomInt(25)));
+                columns.add(new ColumnInfo(randomAlphaOfLength(10), randomAlphaOfLength(10), randomAlphaOfLength(10),
+                        randomFrom(JDBCType.values()), randomInt(25)));
             }
         }
 
@@ -80,7 +81,7 @@ public class SqlResponseTests extends AbstractStreamableXContentTestCase<SqlResp
 
         boolean jdbcEnabled = randomBoolean();
         ToXContent.Params params =
-                new ToXContent.MapParams(Collections.singletonMap(SqlResponse.JDBC_ENABLED_PARAM, Boolean.toString(jdbcEnabled)));
+                new ToXContent.MapParams(Collections.singletonMap(JDBC_ENABLED_PARAM, Boolean.toString(jdbcEnabled)));
 
         XContentBuilder builder = testInstance.toXContent(XContentFactory.jsonBuilder(), params);
         Map<String, Object> rootMap = XContentHelper.convertToMap(builder.bytes(), false, builder.contentType()).v2();

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

@@ -28,6 +28,7 @@ import java.util.Collections;
 
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 import static org.elasticsearch.rest.RestRequest.Method.POST;
+import static org.elasticsearch.xpack.sql.plugin.ColumnInfo.JDBC_ENABLED_PARAM;
 
 public class RestSqlAction extends BaseRestHandler {
     private final SqlLicenseChecker sqlLicenseChecker;
@@ -54,7 +55,7 @@ public class RestSqlAction extends BaseRestHandler {
                 public RestResponse buildResponse(SqlResponse response, XContentBuilder builder) throws Exception {
                     // Make sure we only display JDBC-related data if JDBC is enabled
                     ToXContent.Params params = new ToXContent.DelegatingMapParams(
-                            Collections.singletonMap(SqlResponse.JDBC_ENABLED_PARAM, Boolean.toString(sqlLicenseChecker.isJdbcAllowed())),
+                            Collections.singletonMap(JDBC_ENABLED_PARAM, Boolean.toString(sqlLicenseChecker.isJdbcAllowed())),
                             channel.request());
                     response.toXContent(builder, params);
                     return new BytesRestResponse(getStatus(response), builder);

sql/server/src/main/java/org/elasticsearch/xpack/sql/plugin/RestSqlJdbcAction.java

@@ -139,7 +139,7 @@ public class RestSqlJdbcAction extends AbstractSqlProtocolRestAction {
         return channel -> client.execute(SqlAction.INSTANCE, sqlRequest, toActionListener(channel, response -> {
             List<JDBCType> types = new ArrayList<>(response.columns().size());
             List<ColumnInfo> columns = new ArrayList<>(response.columns().size());
-            for (SqlResponse.ColumnInfo info : response.columns()) {
+            for (org.elasticsearch.xpack.sql.plugin.ColumnInfo info : response.columns()) {
                 types.add(info.jdbcType());
                 columns.add(new ColumnInfo(info.name(), info.jdbcType(), EMPTY, EMPTY, EMPTY, EMPTY, info.displaySize()));
             }

sql/server/src/main/java/org/elasticsearch/xpack/sql/plugin/RestSqlListColumnsAction.java

@@ -0,0 +1,41 @@
+/*
+ * 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.client.node.NodeClient;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.rest.BaseRestHandler;
+import org.elasticsearch.rest.RestController;
+import org.elasticsearch.rest.RestRequest;
+import org.elasticsearch.rest.action.RestToXContentListener;
+
+import java.io.IOException;
+
+import static org.elasticsearch.rest.RestRequest.Method.POST;
+
+public class RestSqlListColumnsAction extends BaseRestHandler {
+
+    public RestSqlListColumnsAction(Settings settings, RestController controller) {
+        super(settings);
+        controller.registerHandler(POST, SqlListColumnsAction.REST_ENDPOINT, this);
+    }
+
+    @Override
+    protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
+        SqlListColumnsRequest listColumnsRequest;
+        try (XContentParser parser = request.contentOrSourceParamParser()) {
+            listColumnsRequest = SqlListColumnsRequest.fromXContent(parser);
+        }
+        return channel -> client.executeLocally(SqlListColumnsAction.INSTANCE, listColumnsRequest, new RestToXContentListener<>(channel));
+    }
+
+    @Override
+    public String getName() {
+        return "xpack_sql_list_columns_action";
+    }
+}
+

sql/server/src/main/java/org/elasticsearch/xpack/sql/plugin/RestSqlListTablesAction.java

@@ -0,0 +1,41 @@
+/*
+ * 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.client.node.NodeClient;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.rest.BaseRestHandler;
+import org.elasticsearch.rest.RestController;
+import org.elasticsearch.rest.RestRequest;
+import org.elasticsearch.rest.action.RestToXContentListener;
+
+import java.io.IOException;
+
+import static org.elasticsearch.rest.RestRequest.Method.POST;
+
+public class RestSqlListTablesAction extends BaseRestHandler {
+
+    public RestSqlListTablesAction(Settings settings, RestController controller) {
+        super(settings);
+        controller.registerHandler(POST, SqlListTablesAction.REST_ENDPOINT, this);
+    }
+
+    @Override
+    protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
+        SqlListTablesRequest listTablesRequest;
+        try (XContentParser parser = request.contentOrSourceParamParser()) {
+            listTablesRequest = SqlListTablesRequest.fromXContent(parser);
+        }
+        return channel -> client.executeLocally(SqlListTablesAction.INSTANCE, listTablesRequest, new RestToXContentListener<>(channel));
+    }
+
+    @Override
+    public String getName() {
+        return "xpack_sql_list_tables_action";
+    }
+}
+

sql/server/src/main/java/org/elasticsearch/xpack/sql/plugin/SqlPlugin.java

@@ -67,7 +67,9 @@ public class SqlPlugin implements ActionPlugin {
         return Arrays.asList(new RestSqlAction(settings, restController, sqlLicenseChecker),
                              new SqlTranslateAction.RestAction(settings, restController),
                              new RestSqlJdbcAction(settings, restController, sqlLicenseChecker, indexResolver),
-                             new RestSqlClearCursorAction(settings, restController));
+                             new RestSqlClearCursorAction(settings, restController),
+                             new RestSqlListTablesAction(settings, restController),
+                             new RestSqlListColumnsAction(settings, restController));
     }
 
     @Override
@@ -78,6 +80,8 @@ public class SqlPlugin implements ActionPlugin {
 
         return Arrays.asList(new ActionHandler<>(SqlAction.INSTANCE, TransportSqlAction.class),
                              new ActionHandler<>(SqlTranslateAction.INSTANCE, SqlTranslateAction.TransportAction.class),
-                             new ActionHandler<>(SqlClearCursorAction.INSTANCE, TransportSqlClearCursorAction.class));
+                             new ActionHandler<>(SqlClearCursorAction.INSTANCE, TransportSqlClearCursorAction.class),
+                             new ActionHandler<>(SqlListTablesAction.INSTANCE, TransportSqlListTablesAction.class),
+                             new ActionHandler<>(SqlListColumnsAction.INSTANCE, TransportSqlListColumnsAction.class));
     }
 }

sql/server/src/main/java/org/elasticsearch/xpack/sql/plugin/TransportSqlAction.java

@@ -16,7 +16,6 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.sql.execution.PlanExecutor;
-import org.elasticsearch.xpack.sql.plugin.SqlResponse.ColumnInfo;
 import org.elasticsearch.xpack.sql.session.Configuration;
 import org.elasticsearch.xpack.sql.session.Cursor;
 import org.elasticsearch.xpack.sql.session.RowSet;
@@ -71,7 +70,7 @@ public class TransportSqlAction extends HandledTransportAction<SqlRequest, SqlRe
     static SqlResponse createResponse(SchemaRowSet rowSet) {
         List<ColumnInfo> columns = new ArrayList<>(rowSet.columnCount());
         for (Schema.Entry entry : rowSet.schema()) {
-            columns.add(new ColumnInfo(entry.name(), entry.type().esName(), entry.type().sqlType(), entry.type().displaySize()));
+            columns.add(new ColumnInfo("", entry.name(), entry.type().esName(), entry.type().sqlType(), entry.type().displaySize()));
         }
         columns = unmodifiableList(columns);
         return createResponse(rowSet, columns);

sql/server/src/main/java/org/elasticsearch/xpack/sql/plugin/TransportSqlListColumnsAction.java

@@ -0,0 +1,67 @@
+/*
+ * 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.action.support.ActionFilters;
+import org.elasticsearch.action.support.HandledTransportAction;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.xpack.sql.analysis.index.EsIndex;
+import org.elasticsearch.xpack.sql.analysis.index.IndexResolver;
+import org.elasticsearch.xpack.sql.type.DataType;
+import org.elasticsearch.xpack.sql.util.StringUtils;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Pattern;
+
+import static org.elasticsearch.common.Strings.hasText;
+
+public class TransportSqlListColumnsAction extends HandledTransportAction<SqlListColumnsRequest, SqlListColumnsResponse> {
+    private final SqlLicenseChecker sqlLicenseChecker;
+    private final IndexResolver indexResolver;
+
+    @Inject
+    public TransportSqlListColumnsAction(Settings settings, ThreadPool threadPool,
+                                         TransportService transportService, ActionFilters actionFilters,
+                                         IndexNameExpressionResolver indexNameExpressionResolver,
+                                         SqlLicenseChecker sqlLicenseChecker, IndexResolver indexResolver) {
+        super(settings, SqlListColumnsAction.NAME, threadPool, transportService, actionFilters, SqlListColumnsRequest::new,
+                indexNameExpressionResolver);
+        this.sqlLicenseChecker = sqlLicenseChecker;
+        this.indexResolver = indexResolver;
+    }
+
+    @Override
+    protected void doExecute(SqlListColumnsRequest request, ActionListener<SqlListColumnsResponse> listener) {
+        sqlLicenseChecker.checkIfSqlAllowed();
+        // TODO: This is wrong
+        // See https://github.com/elastic/x-pack-elasticsearch/pull/3438/commits/61b7c26fe08db2721f0431579f215fe493744af3
+        // and https://github.com/elastic/x-pack-elasticsearch/issues/3460
+        String indexPattern = Strings.hasText(request.getTablePattern()) ? StringUtils.jdbcToEsPattern(request.getTablePattern()) : "*";
+        Pattern columnMatcher = hasText(request.getColumnPattern()) ? StringUtils.likeRegex(request.getColumnPattern()) : null;
+        indexResolver.asList(indexPattern, ActionListener.wrap(esIndices -> {
+            List<ColumnInfo> columns = new ArrayList<>();
+            for (EsIndex esIndex : esIndices) {
+                for (Map.Entry<String, DataType> entry : esIndex.mapping().entrySet()) {
+                    String name = entry.getKey();
+                    if (columnMatcher == null || columnMatcher.matcher(name).matches()) {
+                        DataType type = entry.getValue();
+                        // the column size it's actually its precision (based on the Javadocs)
+                        columns.add(new ColumnInfo(esIndex.name(), name, type.esName(), type.sqlType(), type.displaySize()));
+                    }
+                }
+            }
+            listener.onResponse(new SqlListColumnsResponse(columns));
+        }, listener::onFailure));
+    }
+}

sql/server/src/main/java/org/elasticsearch/xpack/sql/plugin/TransportSqlListTablesAction.java

@@ -0,0 +1,50 @@
+/*
+ * 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.action.support.ActionFilters;
+import org.elasticsearch.action.support.HandledTransportAction;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.xpack.sql.analysis.index.EsIndex;
+import org.elasticsearch.xpack.sql.analysis.index.IndexResolver;
+import org.elasticsearch.xpack.sql.util.StringUtils;
+
+import static java.util.stream.Collectors.toList;
+import static org.elasticsearch.common.Strings.hasText;
+
+public class TransportSqlListTablesAction extends HandledTransportAction<SqlListTablesRequest, SqlListTablesResponse> {
+    private final SqlLicenseChecker sqlLicenseChecker;
+    private final IndexResolver indexResolver;
+
+    @Inject
+    public TransportSqlListTablesAction(Settings settings, ThreadPool threadPool,
+                                        TransportService transportService, ActionFilters actionFilters,
+                                        IndexNameExpressionResolver indexNameExpressionResolver,
+                                        SqlLicenseChecker sqlLicenseChecker, IndexResolver indexResolver) {
+        super(settings, SqlListTablesAction.NAME, threadPool, transportService, actionFilters, SqlListTablesRequest::new,
+                indexNameExpressionResolver);
+        this.sqlLicenseChecker = sqlLicenseChecker;
+        this.indexResolver = indexResolver;
+    }
+
+    @Override
+    protected void doExecute(SqlListTablesRequest request, ActionListener<SqlListTablesResponse> listener) {
+        sqlLicenseChecker.checkIfSqlAllowed();
+        // TODO: This is wrong
+        // See https://github.com/elastic/x-pack-elasticsearch/pull/3438/commits/61b7c26fe08db2721f0431579f215fe493744af3
+        // and https://github.com/elastic/x-pack-elasticsearch/issues/3460
+        String indexPattern = hasText(request.getPattern()) ? StringUtils.jdbcToEsPattern(request.getPattern()) : "*";
+        indexResolver.asList(indexPattern, ActionListener.wrap(list -> listener.onResponse(
+                new SqlListTablesResponse(list.stream()
+                .map(EsIndex::name)
+                .collect(toList()))), listener::onFailure));
+    }
+}

sql/server/src/test/java/org/elasticsearch/xpack/sql/execution/search/CursorTests.java

@@ -13,6 +13,7 @@ import org.elasticsearch.client.Client;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xpack.sql.plugin.CliFormatter;
 import org.elasticsearch.xpack.sql.plugin.CliFormatterCursor;
+import org.elasticsearch.xpack.sql.plugin.ColumnInfo;
 import org.elasticsearch.xpack.sql.plugin.JdbcCursor;
 import org.elasticsearch.xpack.sql.plugin.SqlResponse;
 import org.elasticsearch.xpack.sql.session.Configuration;
@@ -60,12 +61,12 @@ public class CursorTests extends ESTestCase {
     private static SqlResponse createRandomSqlResponse() {
         int columnCount = between(1, 10);
 
-        List<SqlResponse.ColumnInfo> columns = null;
+        List<ColumnInfo> columns = null;
         if (randomBoolean()) {
             columns = new ArrayList<>(columnCount);
             for (int i = 0; i < columnCount; i++) {
-                columns.add(new SqlResponse.ColumnInfo(randomAlphaOfLength(10), randomAlphaOfLength(10),
+                columns.add(new ColumnInfo(randomAlphaOfLength(10), randomAlphaOfLength(10), randomAlphaOfLength(10),
-                    randomFrom(JDBCType.values()), randomInt(25)));
+                        randomFrom(JDBCType.values()), randomInt(25)));
             }
         }
         return new SqlResponse("", columns, Collections.emptyList());

sql/server/src/test/java/org/elasticsearch/xpack/sql/plugin/CliFormatterTests.java

@@ -6,7 +6,6 @@
 package org.elasticsearch.xpack.sql.plugin;
 
 import org.elasticsearch.test.ESTestCase;
-import org.elasticsearch.xpack.sql.plugin.SqlResponse.ColumnInfo;
 
 import java.sql.JDBCType;
 import java.util.Arrays;
@@ -16,11 +15,11 @@ import static org.hamcrest.Matchers.arrayWithSize;
 public class CliFormatterTests extends ESTestCase {
     private final SqlResponse firstResponse = new SqlResponse("",
             Arrays.asList(
-                    new ColumnInfo("foo", "string", JDBCType.VARCHAR, 0),
+                    new ColumnInfo("", "foo", "string", JDBCType.VARCHAR, 0),
-                    new ColumnInfo("bar", "long", JDBCType.BIGINT, 15),
+                    new ColumnInfo("", "bar", "long", JDBCType.BIGINT, 15),
-                    new ColumnInfo("15charwidename!", "double", JDBCType.DOUBLE, 25),
+                    new ColumnInfo("", "15charwidename!", "double", JDBCType.DOUBLE, 25),
-                    new ColumnInfo("superduperwidename!!!", "double", JDBCType.DOUBLE, 25),
+                    new ColumnInfo("", "superduperwidename!!!", "double", JDBCType.DOUBLE, 25),
-                    new ColumnInfo("baz", "keyword", JDBCType.VARCHAR, 0)),
+                    new ColumnInfo("", "baz", "keyword", JDBCType.VARCHAR, 0)),
             Arrays.asList(
                 Arrays.asList("15charwidedata!", 1, 6.888, 12, "rabbit"),
                 Arrays.asList("dog", 1.7976931348623157E308, 123124.888, 9912, "goat")));