SQL: transfer version compatibility decision to the server (#53082)

This commit adds a new request object field, "version", containing the version of the requesting client. This parameter is now accepted - and for certain clients required - by the server and the request is validated against it. Currently server's and client's versions still need to be equal in order for the request to be accepted. Relaxing this check is going to be part of future work. 

On the clients' side, the only check remaining is to ensure that the peer server is supporting version backwards compatibility (i.e. is on, or newer than a certain release).

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

@@ -5,9 +5,10 @@
  */
 package org.elasticsearch.xpack.sql.jdbc;
 
+import org.elasticsearch.xpack.sql.client.ClientVersion;
 import org.elasticsearch.xpack.sql.client.ConnectionConfiguration;
-import org.elasticsearch.xpack.sql.client.Version;
 
+import javax.sql.DataSource;
 import java.io.PrintWriter;
 import java.sql.Connection;
 import java.sql.SQLException;
@@ -16,8 +17,6 @@ import java.sql.Wrapper;
 import java.util.Properties;
 import java.util.logging.Logger;
 
-import javax.sql.DataSource;
-
 /**
  * Factory for connections to Elasticsearch SQL.
  */
@@ -25,7 +24,7 @@ public class EsDataSource implements DataSource, Wrapper {
 
     static {
         // invoke Version to perform classpath/jar sanity checks
-        Version.CURRENT.toString();
+        ClientVersion.CURRENT.toString();
     }
 
     private String url;

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

@@ -5,7 +5,7 @@
  */
 package org.elasticsearch.xpack.sql.jdbc;
 
-import org.elasticsearch.xpack.sql.client.Version;
+import org.elasticsearch.xpack.sql.client.ClientVersion;
 
 import java.io.PrintWriter;
 import java.sql.Connection;
@@ -23,7 +23,7 @@ public class EsDriver implements Driver {
 
     static {
         // invoke Version to perform classpath/jar sanity checks
-        Version.CURRENT.toString();
+        ClientVersion.CURRENT.toString();
 
         try {
             register();
@@ -96,12 +96,12 @@ public class EsDriver implements Driver {
 
     @Override
     public int getMajorVersion() {
-        return Version.CURRENT.major;
+        return ClientVersion.CURRENT.major;
     }
 
     @Override
     public int getMinorVersion() {
-        return Version.CURRENT.minor;
+        return ClientVersion.CURRENT.minor;
     }
 
     @Override

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

@@ -5,28 +5,23 @@
  */
 package org.elasticsearch.xpack.sql.jdbc;
 
+
+import org.elasticsearch.xpack.sql.proto.SqlVersion;
+
 /**
  * General information about the server.
  */
 class InfoResponse {
     final String cluster;
-    final int majorVersion;
-    final int minorVersion;
-    final int revisionVersion;
+    final SqlVersion version;
 
-    InfoResponse(String clusterName, byte versionMajor, byte versionMinor, byte revisionVersion) {
+    InfoResponse(String clusterName, SqlVersion version) {
         this.cluster = clusterName;
-        this.majorVersion = versionMajor;
-        this.minorVersion = versionMinor;
-        this.revisionVersion = revisionVersion;
+        this.version = version;
     }
 
     @Override
     public String toString() {
-        return cluster + "[" + versionString() + "]";
-    }
-    
-    public String versionString() {
-        return majorVersion + "." + minorVersion + "." + revisionVersion;
+        return cluster + "[" + version.toString() + "]";
     }
 }

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

@@ -5,9 +5,9 @@
  */
 package org.elasticsearch.xpack.sql.jdbc;
 
+import org.elasticsearch.xpack.sql.client.ClientVersion;
 import org.elasticsearch.xpack.sql.client.ConnectionConfiguration;
 import org.elasticsearch.xpack.sql.client.StringUtils;
-import org.elasticsearch.xpack.sql.client.Version;
 
 import java.net.URI;
 import java.sql.DriverPropertyInfo;
@@ -70,7 +70,7 @@ public class JdbcConfiguration extends ConnectionConfiguration {
         // typically this should have already happened but in case the
         // EsDriver/EsDataSource are not used and the impl. classes used directly
         // this covers that case
-        Version.CURRENT.toString();
+        ClientVersion.CURRENT.toString();
     }
 
     // immutable properties

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

@@ -426,10 +426,10 @@ class JdbcConnection implements Connection, JdbcWrapper {
     // in fact, this information is cached by the underlying client
     // once retrieved
     int esInfoMajorVersion() throws SQLException {
-        return client.serverInfo().majorVersion;
+        return client.serverInfo().version.major;
     }
 
     int esInfoMinorVersion() throws SQLException {
-        return client.serverInfo().minorVersion;
+        return client.serverInfo().version.minor;
     }
 }

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

@@ -5,8 +5,8 @@
  */
 package org.elasticsearch.xpack.sql.jdbc;
 
+import org.elasticsearch.xpack.sql.client.ClientVersion;
 import org.elasticsearch.xpack.sql.client.ObjectUtils;
-import org.elasticsearch.xpack.sql.client.Version;
 
 import java.sql.Connection;
 import java.sql.DatabaseMetaData;
@@ -94,7 +94,7 @@ class JdbcDatabaseMetaData implements DatabaseMetaData, JdbcWrapper {
 
     @Override
     public String getDatabaseProductVersion() throws SQLException {
-        return Version.CURRENT.toString();
+        return ClientVersion.CURRENT.toString();
     }
 
     @Override
@@ -104,17 +104,17 @@ class JdbcDatabaseMetaData implements DatabaseMetaData, JdbcWrapper {
 
     @Override
     public String getDriverVersion() throws SQLException {
-        return Version.CURRENT.major + "." + Version.CURRENT.minor;
+        return ClientVersion.CURRENT.major + "." + ClientVersion.CURRENT.minor;
     }
 
     @Override
     public int getDriverMajorVersion() {
-        return Version.CURRENT.major;
+        return ClientVersion.CURRENT.major;
     }
 
     @Override
     public int getDriverMinorVersion() {
-        return Version.CURRENT.minor;
+        return ClientVersion.CURRENT.minor;
     }
 
     @Override
@@ -1111,12 +1111,12 @@ class JdbcDatabaseMetaData implements DatabaseMetaData, JdbcWrapper {
 
     @Override
     public int getJDBCMajorVersion() throws SQLException {
-        return Version.jdbcMajorVersion();
+        return ClientVersion.jdbcMajorVersion();
     }
 
     @Override
     public int getJDBCMinorVersion() throws SQLException {
-        return Version.jdbcMinorVersion();
+        return ClientVersion.jdbcMinorVersion();
     }
 
     @Override

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

@@ -7,8 +7,8 @@ package org.elasticsearch.xpack.sql.jdbc;
 
 import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.xpack.sql.client.ClientVersion;
 import org.elasticsearch.xpack.sql.client.HttpClient;
-import org.elasticsearch.xpack.sql.client.Version;
 import org.elasticsearch.xpack.sql.proto.ColumnInfo;
 import org.elasticsearch.xpack.sql.proto.MainResponse;
 import org.elasticsearch.xpack.sql.proto.Mode;
@@ -16,6 +16,7 @@ import org.elasticsearch.xpack.sql.proto.RequestInfo;
 import org.elasticsearch.xpack.sql.proto.SqlQueryRequest;
 import org.elasticsearch.xpack.sql.proto.SqlQueryResponse;
 import org.elasticsearch.xpack.sql.proto.SqlTypedParamValue;
+import org.elasticsearch.xpack.sql.proto.SqlVersion;
 
 import java.sql.SQLException;
 import java.util.ArrayList;
@@ -62,7 +63,7 @@ class JdbcHttpClient {
                 null,
                 Boolean.FALSE,
                 null,
-                new RequestInfo(Mode.JDBC),
+                new RequestInfo(Mode.JDBC, ClientVersion.CURRENT),
                 conCfg.fieldMultiValueLeniency(),
                 conCfg.indexIncludeFrozen(),
                 conCfg.binaryCommunication());
@@ -94,16 +95,15 @@ class JdbcHttpClient {
 
     private InfoResponse fetchServerInfo() throws SQLException {
         MainResponse mainResponse = httpClient.serverInfo();
-        Version version = Version.fromString(mainResponse.getVersion());
-        return new InfoResponse(mainResponse.getClusterName(), version.major, version.minor, version.revision);
+        SqlVersion version = SqlVersion.fromString(mainResponse.getVersion());
+        return new InfoResponse(mainResponse.getClusterName(), version);
     }
-    
+
     private void checkServerVersion() throws SQLException {
-        if (serverInfo.majorVersion != Version.CURRENT.major
-                || serverInfo.minorVersion != Version.CURRENT.minor
-                || serverInfo.revisionVersion != Version.CURRENT.revision) {
+        if (ClientVersion.isServerCompatible(serverInfo.version) == false) {
             throw new SQLException("This version of the JDBC driver is only compatible with Elasticsearch version " +
-                    Version.CURRENT.toString() + ", attempting to connect to a server version " + serverInfo.versionString());
+                ClientVersion.CURRENT.majorMinorToString() + " or newer; attempting to connect to a server version " +
+                serverInfo.version.toString());
         }
     }
 

x-pack/plugin/sql/jdbc/src/test/java/org/elasticsearch/xpack/sql/jdbc/VersionParityTests.java

@@ -12,6 +12,7 @@ import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.test.VersionUtils;
 import org.elasticsearch.test.http.MockResponse;
+import org.elasticsearch.xpack.sql.client.ClientVersion;
 
 import java.io.IOException;
 import java.sql.SQLException;
@@ -31,9 +32,9 @@ public class VersionParityTests extends WebServerTestCase {
         
         String url = JdbcConfiguration.URL_PREFIX + webServerAddress();
         SQLException ex = expectThrows(SQLException.class, () -> new JdbcHttpClient(JdbcConfiguration.create(url, null, 0)));
-        assertEquals("This version of the JDBC driver is only compatible with Elasticsearch version "
-                + org.elasticsearch.xpack.sql.client.Version.CURRENT.toString()
-                + ", attempting to connect to a server version " + version.toString(), ex.getMessage());
+        assertEquals("This version of the JDBC driver is only compatible with Elasticsearch version " +
+            ClientVersion.CURRENT.majorMinorToString() + " or newer; attempting to connect to a server " +
+            "version " + version.toString(), ex.getMessage());
     }
     
     public void testNoExceptionThrownForCompatibleVersions() throws IOException {

x-pack/plugin/sql/jdbc/src/test/java/org/elasticsearch/xpack/sql/jdbc/VersionTests.java

@@ -6,15 +6,14 @@
 package org.elasticsearch.xpack.sql.jdbc;
 
 import org.elasticsearch.test.ESTestCase;
-import org.elasticsearch.xpack.sql.client.Version;
+import org.elasticsearch.xpack.sql.client.ClientVersion;
 
 public class VersionTests extends ESTestCase {
     public void testVersionIsCurrent() {
         /* This test will only work properly in gradle because in gradle we run the tests
          * using the jar. */
-        assertNotNull(Version.CURRENT.hash);
-        assertEquals(org.elasticsearch.Version.CURRENT.major, Version.CURRENT.major);
-        assertEquals(org.elasticsearch.Version.CURRENT.minor, Version.CURRENT.minor);
-        assertEquals(org.elasticsearch.Version.CURRENT.revision, Version.CURRENT.revision);
+        assertEquals(org.elasticsearch.Version.CURRENT.major, ClientVersion.CURRENT.major);
+        assertEquals(org.elasticsearch.Version.CURRENT.minor, ClientVersion.CURRENT.minor);
+        assertEquals(org.elasticsearch.Version.CURRENT.revision, ClientVersion.CURRENT.revision);
     }
 }

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

@@ -26,6 +26,7 @@ import static java.util.Collections.singletonList;
 import static org.elasticsearch.xpack.sql.qa.rest.BaseRestSqlTestCase.mode;
 import static org.elasticsearch.xpack.sql.qa.rest.BaseRestSqlTestCase.randomMode;
 import static org.elasticsearch.xpack.sql.qa.rest.BaseRestSqlTestCase.toMap;
+import static org.elasticsearch.xpack.sql.qa.rest.BaseRestSqlTestCase.version;
 import static org.elasticsearch.xpack.sql.qa.rest.RestSqlTestCase.SQL_QUERY_REST_ENDPOINT;
 import static org.elasticsearch.xpack.sql.qa.rest.RestSqlTestCase.columnInfo;
 
@@ -110,7 +111,7 @@ public class RestSqlMultinodeIT extends ESRestTestCase {
         expected.put("rows", singletonList(singletonList(count)));
 
         Request request = new Request("POST", SQL_QUERY_REST_ENDPOINT);
-        request.setJsonEntity("{\"query\": \"SELECT COUNT(*) FROM test\"" + mode(mode) + "}");
+        request.setJsonEntity("{\"query\": \"SELECT COUNT(*) FROM test\"" + mode(mode) + version(mode) + "}");
         Map<String, Object> actual = toMap(client.performRequest(request), mode);
 
         if (false == expected.equals(actual)) {

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

@@ -32,6 +32,7 @@ import java.util.stream.Collectors;
 
 import static org.elasticsearch.xpack.sql.qa.rest.BaseRestSqlTestCase.mode;
 import static org.elasticsearch.xpack.sql.qa.rest.BaseRestSqlTestCase.randomMode;
+import static org.elasticsearch.xpack.sql.qa.rest.BaseRestSqlTestCase.version;
 import static org.elasticsearch.xpack.sql.qa.rest.RestSqlTestCase.SQL_QUERY_REST_ENDPOINT;
 import static org.elasticsearch.xpack.sql.qa.rest.RestSqlTestCase.columnInfo;
 import static org.hamcrest.Matchers.containsString;
@@ -70,10 +71,10 @@ public class RestSqlSecurityIT extends SqlSecurityTestCase {
         public void expectScrollMatchesAdmin(String adminSql, String user, String userSql) throws Exception {
             String mode = randomMode();
             Map<String, Object> adminResponse = runSql(null,
-                    new StringEntity("{\"query\": \"" + adminSql + "\", \"fetch_size\": 1" + mode(mode) + "}",
+                    new StringEntity("{\"query\": \"" + adminSql + "\", \"fetch_size\": 1" + mode(mode) + version(mode) + "}",
                             ContentType.APPLICATION_JSON), mode);
             Map<String, Object> otherResponse = runSql(user,
-                    new StringEntity("{\"query\": \"" + adminSql + "\", \"fetch_size\": 1" + mode(mode) + "}",
+                    new StringEntity("{\"query\": \"" + adminSql + "\", \"fetch_size\": 1" + mode(mode) + version(mode) + "}",
                             ContentType.APPLICATION_JSON), mode);
 
             String adminCursor = (String) adminResponse.remove("cursor");
@@ -83,9 +84,11 @@ public class RestSqlSecurityIT extends SqlSecurityTestCase {
             assertResponse(adminResponse, otherResponse);
             while (true) {
                 adminResponse = runSql(null,
-                        new StringEntity("{\"cursor\": \"" + adminCursor + "\"" + mode(mode) + "}", ContentType.APPLICATION_JSON), mode);
+                        new StringEntity("{\"cursor\": \"" + adminCursor + "\"" + mode(mode) + version(mode) + "}",
+                            ContentType.APPLICATION_JSON), mode);
                 otherResponse = runSql(user,
-                        new StringEntity("{\"cursor\": \"" + otherCursor + "\"" + mode(mode) + "}", ContentType.APPLICATION_JSON), mode);
+                        new StringEntity("{\"cursor\": \"" + otherCursor + "\"" + mode(mode) + version(mode) + "}",
+                            ContentType.APPLICATION_JSON), mode);
                 adminCursor = (String) adminResponse.remove("cursor");
                 otherCursor = (String) otherResponse.remove("cursor");
                 assertResponse(adminResponse, otherResponse);
@@ -180,7 +183,8 @@ public class RestSqlSecurityIT extends SqlSecurityTestCase {
         }
 
         private static Map<String, Object> runSql(@Nullable String asUser, String mode, String sql) throws IOException {
-            return runSql(asUser, new StringEntity("{\"query\": \"" + sql + "\"" + mode(mode) + "}", ContentType.APPLICATION_JSON), mode);
+            return runSql(asUser, new StringEntity("{\"query\": \"" + sql + "\"" + mode(mode) + version(mode) + "}",
+                ContentType.APPLICATION_JSON), mode);
         }
 
         private static Map<String, Object> runSql(@Nullable String asUser, HttpEntity entity, String mode) throws IOException {
@@ -230,18 +234,18 @@ public class RestSqlSecurityIT extends SqlSecurityTestCase {
      */
     public void testHijackScrollFails() throws Exception {
         createUser("full_access", "rest_minimal");
+        final String mode = randomMode();
 
-        String mode = randomMode();
         Map<String, Object> adminResponse = RestActions.runSql(null,
-                new StringEntity("{\"query\": \"SELECT * FROM test\", \"fetch_size\": 1" + mode(mode) + "}",
+                new StringEntity("{\"query\": \"SELECT * FROM test\", \"fetch_size\": 1" + mode(mode) + version(mode) + "}",
                         ContentType.APPLICATION_JSON), mode);
 
         String cursor = (String) adminResponse.remove("cursor");
         assertNotNull(cursor);
 
-        final String m = randomMode();
         ResponseException e = expectThrows(ResponseException.class, () -> RestActions.runSql("full_access",
-                new StringEntity("{\"cursor\":\"" + cursor + "\"" + mode(m) + "}", ContentType.APPLICATION_JSON), m));
+                new StringEntity("{\"cursor\":\"" + cursor + "\"" + mode(mode) + version(mode) + "}", ContentType.APPLICATION_JSON),
+                mode));
         // TODO return a better error message for bad scrolls
         assertThat(e.getMessage(), containsString("No search context found for id"));
         assertEquals(404, e.getResponse().getStatusLine().getStatusCode());

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

@@ -33,6 +33,7 @@ import java.util.Map;
 import static org.elasticsearch.xpack.sql.qa.rest.BaseRestSqlTestCase.mode;
 import static org.elasticsearch.xpack.sql.qa.rest.BaseRestSqlTestCase.randomMode;
 import static org.elasticsearch.xpack.sql.qa.rest.BaseRestSqlTestCase.toMap;
+import static org.elasticsearch.xpack.sql.qa.rest.BaseRestSqlTestCase.version;
 import static org.elasticsearch.xpack.sql.qa.rest.RestSqlTestCase.SQL_QUERY_REST_ENDPOINT;
 import static org.elasticsearch.xpack.sql.qa.rest.RestSqlTestCase.columnInfo;
 
@@ -177,7 +178,8 @@ public class UserFunctionIT extends ESRestTestCase {
             options.addHeader("es-security-runas-user", asUser);
             request.setOptions(options);
         }
-        request.setEntity(new StringEntity("{\"query\": \"" + sql + "\"" + mode(mode) + "}", ContentType.APPLICATION_JSON));
+        request.setEntity(new StringEntity("{\"query\": \"" + sql + "\"" + mode(mode) + version(mode) + "}",
+            ContentType.APPLICATION_JSON));
         return toMap(client().performRequest(request), mode);
     }
     

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

@@ -42,16 +42,16 @@ public class RestSqlIT extends RestSqlTestCase {
 
     public void testErrorMessageForInvalidParamDataType() throws IOException {
         expectBadRequest(() -> runTranslateSql(
-            "{\"query\":\"SELECT null WHERE 0 = ? \", \"mode\": \"odbc\", \"params\":[{\"type\":\"invalid\", \"value\":\"irrelevant\"}]}"
-            ),
+            "{\"query\":\"SELECT null WHERE 0 = ? \"" + mode("odbc") + version("odbc") +
+                ", \"params\":[{\"type\":\"invalid\", \"value\":\"irrelevant\"}]}"),
             containsString("Invalid parameter data type [invalid]")
         );
     }
 
     public void testErrorMessageForInvalidParamSpec() throws IOException {
         expectBadRequest(() -> runTranslateSql(
-            "{\"query\":\"SELECT null WHERE 0 = ? \", \"mode\": \"odbc\", \"params\":[{\"type\":\"SHAPE\", \"value\":false}]}"
-            ),
+            "{\"query\":\"SELECT null WHERE 0 = ? \"" + mode("odbc") + version("odbc") +
+                ", \"params\":[{\"type\":\"SHAPE\", \"value\":false}]}"),
             containsString("Cannot cast value [false] of type [BOOLEAN] to parameter type [SHAPE]")
         );
     }

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

@@ -31,6 +31,7 @@ import java.util.Map;
 import static org.elasticsearch.xpack.sql.proto.Mode.CLI;
 import static org.elasticsearch.xpack.sql.proto.Protocol.SQL_QUERY_REST_ENDPOINT;
 import static org.elasticsearch.xpack.sql.proto.RequestInfo.CLIENT_IDS;
+import static org.elasticsearch.xpack.sql.qa.rest.BaseRestSqlTestCase.version;
 import static org.elasticsearch.xpack.sql.qa.rest.RestSqlTestCase.mode;
 
 public abstract class SqlProtocolTestCase extends ESRestTestCase {
@@ -135,7 +136,7 @@ public abstract class SqlProtocolTestCase extends ESRestTestCase {
                 + "CAST(-1234.34 AS REAL) AS float_negative,"
                 + "1234567890123.34 AS double_positive,"
                 + "-1234567890123.34 AS double_negative\""
-                + mode(mode.toString()) + "}";
+                + mode(mode.toString()) + version(mode.toString()) + "}";
         request.setEntity(new StringEntity(requestContent, ContentType.APPLICATION_JSON));
         
         Map<String, Object> map;
@@ -188,7 +189,7 @@ public abstract class SqlProtocolTestCase extends ESRestTestCase {
     private void assertQuery(String sql, String columnName, String columnType, Object columnValue, int displaySize, Mode mode)
             throws IOException {
         boolean columnar = randomBoolean();
-        Map<String, Object> response = runSql(mode.toString(), sql, columnar);
+        Map<String, Object> response = runSql(mode, sql, columnar);
         List<Object> columns = (ArrayList<Object>) response.get("columns");
         assertEquals(1, columns.size());
 
@@ -215,12 +216,12 @@ public abstract class SqlProtocolTestCase extends ESRestTestCase {
             assertEquals(columnValue, row.get(0));
         }
     }
-    
-    private Map<String, Object> runSql(String mode, String sql, boolean columnar) throws IOException {
+
+    private Map<String, Object> runSql(Mode mode, String sql, boolean columnar) throws IOException {
         Request request = new Request("POST", SQL_QUERY_REST_ENDPOINT);
-        String requestContent = "{\"query\":\"" + sql + "\"" + mode(mode) + "}";
+        String requestContent = "{\"query\":\"" + sql + "\"" + mode(mode.toString()) + version(mode.toString()) + "}";
         String format = randomFrom(XContentType.values()).name().toLowerCase(Locale.ROOT);
-        
+
         // add a client_id to the request
         if (randomBoolean()) {
             String clientId = randomFrom(randomFrom(CLIENT_IDS), randomAlphaOfLengthBetween(10, 20));
@@ -252,14 +253,13 @@ public abstract class SqlProtocolTestCase extends ESRestTestCase {
 
         // randomize binary response enforcement for drivers (ODBC/JDBC) and CLI
         boolean binaryCommunication = randomBoolean();
-        Mode m = Mode.fromString(mode);
         if (randomBoolean()) {
             // set it explicitly or leave the default (null) as is
             requestContent = new StringBuilder(requestContent)
                     .insert(requestContent.length() - 1, ",\"binary_format\":" + binaryCommunication).toString();
-            binaryCommunication = ((Mode.isDriver(m) || m == Mode.CLI) && binaryCommunication);
+            binaryCommunication = Mode.isDedicatedClient(mode) && binaryCommunication;
         } else {
-            binaryCommunication = Mode.isDriver(m) || m == Mode.CLI;
+            binaryCommunication = Mode.isDedicatedClient(mode);
         }
         
         // send the query either as body or as request parameter

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

@@ -6,6 +6,7 @@
 
 package org.elasticsearch.xpack.sql.qa.rest;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.Response;
 import org.elasticsearch.common.Strings;
@@ -38,6 +39,14 @@ public abstract class BaseRestSqlTestCase extends ESRestTestCase {
         return Strings.isEmpty(mode) ? StringUtils.EMPTY : ",\"mode\":\"" + mode + "\"";
     }
 
+    public static String version(String mode) {
+        Mode m = Mode.fromString(mode);
+        if (Mode.isDedicatedClient(m)) {
+            return ",\"version\":" + "\"" + Version.CURRENT.toString() + "\"";
+        }
+        return StringUtils.EMPTY;
+    }
+
     public static String randomMode() {
         return randomFrom(StringUtils.EMPTY, "jdbc", "plain");
     }
@@ -49,7 +58,7 @@ public abstract class BaseRestSqlTestCase extends ESRestTestCase {
     public static Number xContentDependentFloatingNumberValue(String mode, Number value) {
         Mode m = Mode.fromString(mode);
         // for drivers and the CLI return the number as is, while for REST cast it implicitly to Double (the JSON standard).
-        if (Mode.isDriver(m) || m == Mode.CLI) {
+        if (Mode.isDedicatedClient(m)) {
             return value;
         } else {
             return value.doubleValue();

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

@@ -105,8 +105,9 @@ public abstract class RestSqlTestCase extends BaseRestSqlTestCase implements Err
                 + "   SELECT text, number, SQRT(number) AS s, SCORE()"
                 + "     FROM test"
                 + " ORDER BY number, SCORE()\", "
-                + "\"mode\":\"" + mode + "\", "
-            + "\"fetch_size\":2" + columnarParameter(columnar) + "}";
+                + "\"mode\":\"" + mode + "\""
+                + version(mode)
+                + ", \"fetch_size\":2" + columnarParameter(columnar) + "}";
 
         Number value = xContentDependentFloatingNumberValue(mode, 1f);
         String cursor = null;
@@ -116,7 +117,8 @@ public abstract class RestSqlTestCase extends BaseRestSqlTestCase implements Err
                 response = runSql(new StringEntity(sqlRequest, ContentType.APPLICATION_JSON), "", mode);
             } else {
                 columnar = randomBoolean();
-                response = runSql(new StringEntity("{\"cursor\":\"" + cursor + "\"" + mode(mode) + columnarParameter(columnar) + "}",
+                response = runSql(new StringEntity("{\"cursor\":\"" + cursor + "\"" + mode(mode) + version(mode) +
+                        columnarParameter(columnar) + "}",
                         ContentType.APPLICATION_JSON), StringUtils.EMPTY, mode);
             }
 
@@ -151,7 +153,8 @@ public abstract class RestSqlTestCase extends BaseRestSqlTestCase implements Err
         } else {
             expected.put("rows", emptyList());
         }
-        assertResponse(expected, runSql(new StringEntity("{ \"cursor\":\"" + cursor + "\"" + mode(mode) + columnarParameter(columnar) + "}",
+        assertResponse(expected, runSql(new StringEntity("{ \"cursor\":\"" + cursor + "\"" + mode(mode) + version(mode) +
+                columnarParameter(columnar) + "}",
                 ContentType.APPLICATION_JSON), StringUtils.EMPTY, mode));
     }
 
@@ -187,8 +190,9 @@ public abstract class RestSqlTestCase extends BaseRestSqlTestCase implements Err
         String sqlRequest =
                 "{\"query\":\"SELECT DATE_PART('TZOFFSET', date) AS tz FROM test_date_timezone ORDER BY date\","
                         + "\"time_zone\":\"" + zoneId.getId() + "\", "
-                        + "\"mode\":\"" + mode + "\", "
-                        + "\"fetch_size\":2}";
+                        + "\"mode\":\"" + mode + "\""
+                        + version(mode)
+                        + ",\"fetch_size\":2}";
 
         String cursor = null;
         for (int i = 0; i <= datetimes.length; i += 2) {
@@ -199,7 +203,7 @@ public abstract class RestSqlTestCase extends BaseRestSqlTestCase implements Err
                 expected.put("columns", singletonList(columnInfo(mode, "tz", "integer", JDBCType.INTEGER, 11)));
                 response = runSql(new StringEntity(sqlRequest, ContentType.APPLICATION_JSON), "", mode);
             } else {
-                response = runSql(new StringEntity("{\"cursor\":\"" + cursor + "\"" + mode(mode) + "}",
+                response = runSql(new StringEntity("{\"cursor\":\"" + cursor + "\"" + mode(mode) + version(mode) + "}",
                         ContentType.APPLICATION_JSON), StringUtils.EMPTY, mode);
             }
 
@@ -215,7 +219,7 @@ public abstract class RestSqlTestCase extends BaseRestSqlTestCase implements Err
         }
         Map<String, Object> expected = new HashMap<>();
         expected.put("rows", emptyList());
-        assertResponse(expected, runSql(new StringEntity("{ \"cursor\":\"" + cursor + "\"" + mode(mode) + "}",
+        assertResponse(expected, runSql(new StringEntity("{ \"cursor\":\"" + cursor + "\"" + mode(mode) + version(mode) + "}",
                 ContentType.APPLICATION_JSON), StringUtils.EMPTY, mode));
     }
 
@@ -423,9 +427,10 @@ public abstract class RestSqlTestCase extends BaseRestSqlTestCase implements Err
     
     public void testUseColumnarForTranslateRequest() throws IOException {
         index("{\"test\":\"test\"}", "{\"test\":\"test\"}");
-        
+
+        String mode = randomMode();
         Request request = new Request("POST", SQL_TRANSLATE_REST_ENDPOINT);
-        request.setEntity(new StringEntity("{\"columnar\":true,\"query\":\"SELECT * FROM test\"" + mode(randomMode()) + "}",
+        request.setEntity(new StringEntity("{\"columnar\":true,\"query\":\"SELECT * FROM test\"" + mode(mode) + version(mode) + "}",
                 ContentType.APPLICATION_JSON));
         expectBadRequest(() -> {
                 client().performRequest(request);
@@ -464,7 +469,7 @@ public abstract class RestSqlTestCase extends BaseRestSqlTestCase implements Err
 
     private Map<String, Object> runSql(String mode, String sql, String suffix, boolean columnar) throws IOException {
         // put an explicit "columnar": false parameter or omit it altogether, it should make no difference
-        return runSql(new StringEntity("{\"query\":\"" + sql + "\"" + mode(mode) + columnarParameter(columnar) + "}",
+        return runSql(new StringEntity("{\"query\":\"" + sql + "\"" + mode(mode) + version(mode) + columnarParameter(columnar) + "}",
                 ContentType.APPLICATION_JSON), suffix, mode);
     }
     
@@ -567,9 +572,10 @@ public abstract class RestSqlTestCase extends BaseRestSqlTestCase implements Err
             options.addHeader("Accept", randomFrom("*/*", "application/json"));
             request.setOptions(options);
         }
-        request.setEntity(new StringEntity("{\"query\":\"SELECT * FROM test\"" + mode("plain") + columnarParameter(columnar) + "}",
-                  ContentType.APPLICATION_JSON));
-        
+        request.setEntity(new StringEntity("{\"query\":\"SELECT * FROM test\"" + mode("plain") + version("plain") +
+                columnarParameter(columnar) + "}",
+                ContentType.APPLICATION_JSON));
+
         Response response = client().performRequest(request);
         try (InputStream content = response.getEntity().getContent()) {
             String actualJson = new BytesArray(content.readAllBytes()).utf8ToString();
@@ -598,7 +604,7 @@ public abstract class RestSqlTestCase extends BaseRestSqlTestCase implements Err
         expected.put("columns", singletonList(columnInfo(mode, "test", "text", JDBCType.VARCHAR, Integer.MAX_VALUE)));
         expected.put("rows", singletonList(singletonList("foo")));
         assertResponse(expected, runSql(new StringEntity("{\"query\":\"SELECT * FROM test\", " +
-                "\"filter\":{\"match\": {\"test\": \"foo\"}}" + mode(mode) + "}",
+                "\"filter\":{\"match\": {\"test\": \"foo\"}}" + mode(mode) + version(mode) + "}",
                 ContentType.APPLICATION_JSON), StringUtils.EMPTY, mode));
     }
 
@@ -623,7 +629,7 @@ public abstract class RestSqlTestCase extends BaseRestSqlTestCase implements Err
             "10, \"foo\"";
         assertResponse(expected, runSql(new StringEntity("{\"query\":\"SELECT test, ? param FROM test WHERE test = ?\", " +
                 "\"params\":[" + params + "]"
-                + mode(mode) + columnarParameter(columnar) + "}", ContentType.APPLICATION_JSON), StringUtils.EMPTY, mode));
+                + mode(mode) + version(mode) + columnarParameter(columnar) + "}", ContentType.APPLICATION_JSON), StringUtils.EMPTY, mode));
     }
 
     public void testBasicTranslateQueryWithFilter() throws IOException {

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

@@ -29,6 +29,7 @@ import java.util.Map;
 import static org.elasticsearch.xpack.sql.proto.Protocol.SQL_QUERY_REST_ENDPOINT;
 import static org.elasticsearch.xpack.sql.proto.Protocol.SQL_STATS_REST_ENDPOINT;
 import static org.elasticsearch.xpack.sql.proto.Protocol.SQL_TRANSLATE_REST_ENDPOINT;
+import static org.elasticsearch.xpack.sql.qa.rest.BaseRestSqlTestCase.version;
 import static org.elasticsearch.xpack.sql.qa.rest.RestSqlTestCase.mode;
 
 public abstract class RestSqlUsageTestCase extends ESRestTestCase {
@@ -255,18 +256,18 @@ public abstract class RestSqlUsageTestCase extends ESRestTestCase {
     }
     
     private void runSql(String sql) throws IOException {
-        String mode = Mode.PLAIN.toString();
+        Mode mode = Mode.PLAIN;
         if (clientType.equals(ClientType.JDBC.toString())) {
-            mode = Mode.JDBC.toString();
+            mode = Mode.JDBC;
         } else if (clientType.startsWith(ClientType.ODBC.toString())) {
-            mode = Mode.ODBC.toString();
+            mode = Mode.ODBC;
         } else if (clientType.equals(ClientType.CLI.toString())) {
-            mode = Mode.CLI.toString();
+            mode = Mode.CLI;
         }
 
-        runSql(mode, clientType, sql);
+        runSql(mode.toString(), clientType, sql);
     }
-    
+
     @SuppressWarnings({ "unchecked", "rawtypes" })
     private void assertTranslateQueryMetric(int expected, Map<String, Object> responseAsMap) throws IOException {
         List<Map<String, Map<String, Map>>> nodesListStats = (List) responseAsMap.get("stats");
@@ -278,7 +279,7 @@ public abstract class RestSqlUsageTestCase extends ESRestTestCase {
         }
         assertEquals(expected, actualMetricValue);
     }
-    
+
     private void runSql(String mode, String restClient, String sql) throws IOException {
         Request request = new Request("POST", SQL_QUERY_REST_ENDPOINT);
         request.addParameter("error_trace", "true");   // Helps with debugging in case something crazy happens on the server.
@@ -293,9 +294,8 @@ public abstract class RestSqlUsageTestCase extends ESRestTestCase {
             options.addHeader("Accept", randomFrom("*/*", "application/json"));
             request.setOptions(options);
         }
-        request.setEntity(new StringEntity("{\"query\":\"" + sql + "\"" + mode(mode) +
-                (ignoreClientType ? StringUtils.EMPTY : ",\"client_id\":\"" + restClient + "\"") + "}",
-                ContentType.APPLICATION_JSON));
+        request.setEntity(new StringEntity("{\"query\":\"" + sql + "\"" + mode(mode) + version(mode) +
+            (ignoreClientType ? StringUtils.EMPTY : ",\"client_id\":\"" + restClient + "\"") + "}", ContentType.APPLICATION_JSON));
         client().performRequest(request);
     }
     

x-pack/plugin/sql/sql-action/src/main/java/org/elasticsearch/xpack/sql/action/AbstractSqlQueryRequest.java

@@ -5,9 +5,12 @@
  */
 package org.elasticsearch.xpack.sql.action;
 
+import org.elasticsearch.Version;
+import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.CompositeIndicesRequest;
 import org.elasticsearch.common.Nullable;
 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.unit.TimeValue;
@@ -33,6 +36,8 @@ import java.util.List;
 import java.util.Objects;
 import java.util.function.Supplier;
 
+import static org.elasticsearch.action.ValidateActions.addValidationError;
+
 /**
  * Base class for requests that contain sql queries (Query and Translate)
  */
@@ -46,7 +51,8 @@ public abstract class AbstractSqlQueryRequest extends AbstractSqlRequest impleme
     @Nullable
     private QueryBuilder filter = null;
     private List<SqlTypedParamValue> params = Collections.emptyList();
-    
+
+    // TODO: define all REST request object field names in a protocol class as unique source
     static final ParseField QUERY = new ParseField("query");
     static final ParseField CURSOR = new ParseField("cursor");
     static final ParseField PARAMS = new ParseField("params");
@@ -57,6 +63,7 @@ public abstract class AbstractSqlQueryRequest extends AbstractSqlRequest impleme
     static final ParseField FILTER = new ParseField("filter");
     static final ParseField MODE = new ParseField("mode");
     static final ParseField CLIENT_ID = new ParseField("client_id");
+    static final ParseField CLIENT_VERSION = new ParseField("version");
 
     public AbstractSqlQueryRequest() {
         super();
@@ -80,7 +87,8 @@ public abstract class AbstractSqlQueryRequest extends AbstractSqlRequest impleme
         ObjectParser<R, Void> parser = new ObjectParser<>("sql/query", false, supplier);
         parser.declareString(AbstractSqlQueryRequest::query, QUERY);
         parser.declareString((request, mode) -> request.mode(Mode.fromString(mode)), MODE);
-        parser.declareString((request, clientId) -> request.clientId(clientId), CLIENT_ID);
+        parser.declareString(AbstractSqlRequest::clientId, CLIENT_ID);
+        parser.declareString(AbstractSqlRequest::version, CLIENT_VERSION);
         parser.declareField(AbstractSqlQueryRequest::params, AbstractSqlQueryRequest::parseParams, PARAMS, ValueType.VALUE_ARRAY);
         parser.declareString((request, zoneId) -> request.zoneId(ZoneId.of(zoneId)), TIME_ZONE);
         parser.declareInt(AbstractSqlQueryRequest::fetchSize, FETCH_SIZE);
@@ -205,6 +213,26 @@ public abstract class AbstractSqlQueryRequest extends AbstractSqlRequest impleme
         }
     }
 
+    @Override
+    public ActionRequestValidationException validate() {
+        ActionRequestValidationException validationException = null;
+        // the version field is mandatory for drivers and CLI
+        Mode mode = requestInfo().mode();
+        if (mode != null && (Mode.isDedicatedClient(mode))) {
+            if (requestInfo().version() == null) {
+                if (Strings.hasText(query())) {
+                    validationException = addValidationError("[version] is required for the [" + mode.toString() + "] client",
+                        validationException);
+                }
+            } else if (requestInfo().version().equals(Version.CURRENT.toString()) == false) {
+                validationException = addValidationError("The [" + requestInfo().version() + "] version of the [" +
+                        mode.toString() + "] " + "client is not compatible with Elasticsearch version [" + Version.CURRENT + "]",
+                    validationException);
+            }
+        }
+        return validationException;
+    }
+
     /**
      * The client's time zone
      */

x-pack/plugin/sql/sql-action/src/main/java/org/elasticsearch/xpack/sql/action/AbstractSqlRequest.java

@@ -12,6 +12,7 @@ import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.xpack.sql.proto.Mode;
 import org.elasticsearch.xpack.sql.proto.RequestInfo;
+import org.elasticsearch.xpack.sql.proto.SqlVersion;
 
 import java.io.IOException;
 import java.util.Objects;
@@ -39,7 +40,8 @@ public abstract class AbstractSqlRequest extends ActionRequest implements ToXCon
         super(in);
         Mode mode = in.readEnum(Mode.class);
         String clientId = in.readOptionalString();
-        requestInfo = new RequestInfo(mode, clientId);
+        String clientVersion = in.readOptionalString();
+        requestInfo = new RequestInfo(mode, clientId, clientVersion);
     }
 
     @Override
@@ -56,8 +58,9 @@ public abstract class AbstractSqlRequest extends ActionRequest implements ToXCon
         super.writeTo(out);
         out.writeEnum(requestInfo.mode());
         out.writeOptionalString(requestInfo.clientId());
+        out.writeOptionalString(requestInfo.version() == null ? null : requestInfo.version().toString());
     }
-    
+
     public RequestInfo requestInfo() {
         return requestInfo;
     }
@@ -86,6 +89,14 @@ public abstract class AbstractSqlRequest extends ActionRequest implements ToXCon
         this.requestInfo.clientId(clientId);
     }
 
+    public void version(String clientVersion) {
+        requestInfo.version(clientVersion);
+    }
+
+    public SqlVersion version() {
+        return requestInfo.version();
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;

x-pack/plugin/sql/sql-action/src/main/java/org/elasticsearch/xpack/sql/action/SqlClearCursorRequest.java

@@ -21,6 +21,7 @@ import static org.elasticsearch.action.ValidateActions.addValidationError;
 import static org.elasticsearch.common.xcontent.ConstructingObjectParser.constructorArg;
 import static org.elasticsearch.common.xcontent.ConstructingObjectParser.optionalConstructorArg;
 import static org.elasticsearch.xpack.sql.action.AbstractSqlQueryRequest.CLIENT_ID;
+import static org.elasticsearch.xpack.sql.action.AbstractSqlQueryRequest.CLIENT_VERSION;
 import static org.elasticsearch.xpack.sql.action.AbstractSqlQueryRequest.CURSOR;
 import static org.elasticsearch.xpack.sql.action.AbstractSqlQueryRequest.MODE;
 
@@ -42,6 +43,7 @@ public class SqlClearCursorRequest extends AbstractSqlRequest {
         PARSER.declareString(constructorArg(), CURSOR);
         PARSER.declareString(optionalConstructorArg(), MODE);
         PARSER.declareString(optionalConstructorArg(), CLIENT_ID);
+        PARSER.declareString(optionalConstructorArg(), CLIENT_VERSION);
     }
 
     private String cursor;

x-pack/plugin/sql/sql-action/src/main/java/org/elasticsearch/xpack/sql/action/SqlQueryRequest.java

@@ -72,7 +72,7 @@ public class SqlQueryRequest extends AbstractSqlQueryRequest {
 
     @Override
     public ActionRequestValidationException validate() {
-        ActionRequestValidationException validationException = null;
+        ActionRequestValidationException validationException = super.validate();
         if ((false == Strings.hasText(query())) && Strings.hasText(cursor) == false) {
             validationException = addValidationError("one of [query] or [cursor] is required", validationException);
         }

x-pack/plugin/sql/sql-action/src/main/java/org/elasticsearch/xpack/sql/action/SqlQueryRequestBuilder.java

@@ -52,6 +52,11 @@ public class SqlQueryRequestBuilder extends ActionRequestBuilder<SqlQueryRequest
         return this;
     }
 
+    public SqlQueryRequestBuilder version(String version) {
+        request.version(version);
+        return this;
+    }
+
     public SqlQueryRequestBuilder cursor(String cursor) {
         request.cursor(cursor);
         return this;

x-pack/plugin/sql/sql-action/src/main/java/org/elasticsearch/xpack/sql/action/SqlTranslateRequest.java

@@ -44,7 +44,7 @@ public class SqlTranslateRequest extends AbstractSqlQueryRequest {
 
     @Override
     public ActionRequestValidationException validate() {
-        ActionRequestValidationException validationException = null;
+        ActionRequestValidationException validationException = super.validate();
         if ((false == Strings.hasText(query()))) {
             validationException = addValidationError("query is required", validationException);
         }
@@ -65,12 +65,12 @@ public class SqlTranslateRequest extends AbstractSqlQueryRequest {
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         // This is needed just to test parsing of SqlTranslateRequest, so we can reuse SqlQuerySerialization
-        return new SqlQueryRequest(query(), params(), zoneId(), fetchSize(), requestTimeout(), pageTimeout(), 
-            filter(), 
-            null, 
-            null, 
+        return new SqlQueryRequest(query(), params(), zoneId(), fetchSize(), requestTimeout(), pageTimeout(),
+            filter(),
+            null,
+            null,
             requestInfo(),
-            false, 
+            false,
             false,
             null).toXContent(builder, params);
     }

x-pack/plugin/sql/sql-action/src/test/java/org/elasticsearch/xpack/sql/action/SqlQueryRequestTests.java

@@ -176,6 +176,9 @@ public class SqlQueryRequestTests extends AbstractWireSerializingTestCase<SqlQue
         if (request.clientId() != null) {
             builder.field("client_id", request.clientId());
         }
+        if (request.version() != null) {
+            builder.field("version", request.version().toString());
+        }
         if (request.params() != null && request.params().isEmpty() == false) {
             builder.startArray("params");
             for (SqlTypedParamValue val : request.params()) {

x-pack/plugin/sql/sql-action/src/test/java/org/elasticsearch/xpack/sql/action/SqlRequestParsersTests.java

@@ -6,6 +6,7 @@
 
 package org.elasticsearch.xpack.sql.action;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.xcontent.DeprecationHandler;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
@@ -48,9 +49,10 @@ public class SqlRequestParsersTests extends ESTestCase {
         Mode randomMode = randomFrom(Mode.values());
         
         SqlClearCursorRequest request = generateRequest("{\"cursor\" : \"whatever\", \"mode\" : \""
-                + randomMode.toString() + "\", \"client_id\" : \"bla\"}",
+                + randomMode.toString() + "\", \"client_id\" : \"bla\", \"version\": \"1.2.3\"}",
                 SqlClearCursorRequest::fromXContent);
         assertNull(request.clientId());
+        assertNull(request.version());
         assertEquals(randomMode, request.mode());
         assertEquals("whatever", request.getCursor());
         
@@ -66,10 +68,11 @@ public class SqlRequestParsersTests extends ESTestCase {
         assertNull(request.clientId());
         assertEquals(Mode.PLAIN, request.mode());
         assertEquals("whatever", request.getCursor());
-        
-        request = generateRequest("{\"cursor\" : \"whatever\", \"client_id\" : \"CLI\"}",
+
+        request = generateRequest("{\"cursor\" : \"whatever\", \"client_id\" : \"CLI\", \"version\": \"1.2.3\"}",
                 SqlClearCursorRequest::fromXContent);
         assertNull(request.clientId());
+        assertNull(request.version());
         assertEquals(Mode.PLAIN, request.mode());
         assertEquals("whatever", request.getCursor());
         
@@ -101,6 +104,8 @@ public class SqlRequestParsersTests extends ESTestCase {
                 SqlQueryRequest::fromXContent);
         assertParsingErrorMessage("{\"client_id\":123}", "client_id doesn't support values of type: VALUE_NUMBER",
                 SqlQueryRequest::fromXContent);
+        assertParsingErrorMessage("{\"version\":123}", "version doesn't support values of type: VALUE_NUMBER",
+            SqlQueryRequest::fromXContent);
         assertParsingErrorMessage("{\"params\":[{\"value\":123}]}", "failed to parse field [params]", SqlQueryRequest::fromXContent);
         assertParsingErrorMessage("{\"time_zone\":12}", "time_zone doesn't support values of type: VALUE_NUMBER",
                 SqlQueryRequest::fromXContent);
@@ -108,7 +113,10 @@ public class SqlRequestParsersTests extends ESTestCase {
         Mode randomMode = randomFrom(Mode.values());
         String params;
         List<SqlTypedParamValue> list = new ArrayList<>(1);
-        
+
+        final String clientVersion = Mode.isDedicatedClient(randomMode)
+            ? "\"version\": \"" + Version.CURRENT.toString() + "\","
+            : "";
         if (Mode.isDriver(randomMode)) {
             params = "{\"value\":123, \"type\":\"whatever\"}";
             list.add(new SqlTypedParamValue("whatever", 123, true));
@@ -119,12 +127,16 @@ public class SqlRequestParsersTests extends ESTestCase {
         
         SqlQueryRequest request = generateRequest("{\"cursor\" : \"whatever\", \"mode\" : \""
                 + randomMode.toString() + "\", \"client_id\" : \"bla\","
+                + clientVersion
                 + "\"query\":\"select\","
                 + "\"params\":[" + params + "],"
                 + " \"time_zone\":\"UTC\","
                 + "\"request_timeout\":\"5s\",\"page_timeout\":\"10s\"}", SqlQueryRequest::fromXContent);
         assertNull(request.clientId());
         assertEquals(randomMode, request.mode());
+        if (Mode.isDedicatedClient(randomMode)) {
+            assertEquals(Version.CURRENT.toString(), request.version().toString());
+        }
         assertEquals("whatever", request.cursor());
         assertEquals("select", request.query());
 

x-pack/plugin/sql/sql-cli/src/main/java/org/elasticsearch/xpack/sql/cli/Cli.java

@@ -22,9 +22,9 @@ import org.elasticsearch.xpack.sql.cli.command.PrintLogoCommand;
 import org.elasticsearch.xpack.sql.cli.command.ServerInfoCliCommand;
 import org.elasticsearch.xpack.sql.cli.command.ServerQueryCliCommand;
 import org.elasticsearch.xpack.sql.client.ClientException;
+import org.elasticsearch.xpack.sql.client.ClientVersion;
 import org.elasticsearch.xpack.sql.client.ConnectionConfiguration;
 import org.elasticsearch.xpack.sql.client.HttpClient;
-import org.elasticsearch.xpack.sql.client.Version;
 import org.jline.terminal.TerminalBuilder;
 
 import java.io.IOException;
@@ -161,7 +161,7 @@ public class Cli extends LoggingAwareCommand {
                 // Most likely we connected to something other than Elasticsearch
                 throw new UserException(ExitCodes.DATA_ERROR,
                         "Cannot communicate with the server " + con.connectionString() +
-                                ". This version of CLI only works with Elasticsearch version " + Version.CURRENT.toString());
+                                ". This version of CLI only works with Elasticsearch version " + ClientVersion.CURRENT.toString());
             }
         }
     }

x-pack/plugin/sql/sql-cli/src/main/java/org/elasticsearch/xpack/sql/cli/command/CliSession.java

@@ -5,11 +5,12 @@
  */
 package org.elasticsearch.xpack.sql.cli.command;
 
-import org.elasticsearch.xpack.sql.client.HttpClient;
 import org.elasticsearch.xpack.sql.client.ClientException;
-import org.elasticsearch.xpack.sql.client.Version;
+import org.elasticsearch.xpack.sql.client.ClientVersion;
+import org.elasticsearch.xpack.sql.client.HttpClient;
 import org.elasticsearch.xpack.sql.proto.MainResponse;
 import org.elasticsearch.xpack.sql.proto.Protocol;
+import org.elasticsearch.xpack.sql.proto.SqlVersion;
 
 import java.sql.SQLException;
 
@@ -73,11 +74,11 @@ public class CliSession {
         } catch (SQLException ex) {
             throw new ClientException(ex);
         }
-        Version version = Version.fromString(response.getVersion());
-        // TODO: We can relax compatibility requirement later when we have a better idea about protocol compatibility guarantees
-        if (version.major != Version.CURRENT.major || version.minor != Version.CURRENT.minor) {
-            throw new ClientException("This alpha version of CLI is only compatible with Elasticsearch version " +
-                    Version.CURRENT.toString());
+        SqlVersion version = SqlVersion.fromString(response.getVersion());
+        if (ClientVersion.isServerCompatible(version) == false) {
+            throw new ClientException("This version of the CLI is only compatible with Elasticsearch version " +
+                ClientVersion.CURRENT.majorMinorToString() + " or newer; attempting to connect to a server version " +
+                version.toString());
         }
     }
 }

x-pack/plugin/sql/sql-cli/src/main/java/org/elasticsearch/xpack/sql/cli/command/PrintLogoCommand.java

@@ -8,7 +8,7 @@ package org.elasticsearch.xpack.sql.cli.command;
 import org.elasticsearch.xpack.sql.cli.Cli;
 import org.elasticsearch.xpack.sql.cli.CliTerminal;
 import org.elasticsearch.xpack.sql.cli.FatalCliException;
-import org.elasticsearch.xpack.sql.client.Version;
+import org.elasticsearch.xpack.sql.client.ClientVersion;
 
 import java.io.BufferedReader;
 import java.io.IOException;
@@ -53,9 +53,9 @@ public class PrintLogoCommand extends AbstractCliCommand {
         }
 
         // print the version centered on the last line
-        char[] whitespaces = new char[(lineLength - Version.CURRENT.version.length()) / 2];
+        char[] whitespaces = new char[(lineLength - ClientVersion.CURRENT.version.length()) / 2];
         Arrays.fill(whitespaces, ' ');
-        terminal.println(new StringBuilder().append(whitespaces).append(Version.CURRENT.version).toString());
+        terminal.println(new StringBuilder().append(whitespaces).append(ClientVersion.CURRENT.version).toString());
         terminal.println();
     }
 

x-pack/plugin/sql/sql-cli/src/test/java/org/elasticsearch/xpack/sql/cli/CliSessionTests.java

@@ -9,10 +9,11 @@ import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xpack.sql.cli.command.CliSession;
-import org.elasticsearch.xpack.sql.client.HttpClient;
 import org.elasticsearch.xpack.sql.client.ClientException;
-import org.elasticsearch.xpack.sql.client.Version;
+import org.elasticsearch.xpack.sql.client.ClientVersion;
+import org.elasticsearch.xpack.sql.client.HttpClient;
 import org.elasticsearch.xpack.sql.proto.MainResponse;
+import org.elasticsearch.xpack.sql.proto.SqlVersion;
 
 import java.sql.SQLException;
 
@@ -26,7 +27,7 @@ public class CliSessionTests extends ESTestCase {
 
     public void testProperConnection() throws Exception {
         HttpClient httpClient = mock(HttpClient.class);
-        when(httpClient.serverInfo()).thenReturn(new MainResponse(randomAlphaOfLength(5), org.elasticsearch.Version.CURRENT.toString(),
+        when(httpClient.serverInfo()).thenReturn(new MainResponse(randomAlphaOfLength(5), ClientVersion.CURRENT.toString(),
                 ClusterName.DEFAULT.value(), UUIDs.randomBase64UUID()));
         CliSession cliSession = new CliSession(httpClient);
         cliSession.checkConnection();
@@ -38,28 +39,43 @@ public class CliSessionTests extends ESTestCase {
         HttpClient httpClient = mock(HttpClient.class);
         when(httpClient.serverInfo()).thenThrow(new SQLException("Cannot connect"));
         CliSession cliSession = new CliSession(httpClient);
-        expectThrows(ClientException.class, cliSession::checkConnection);
+        Throwable throwable = expectThrows(ClientException.class, cliSession::checkConnection);
+        assertEquals("java.sql.SQLException: Cannot connect", throwable.getMessage());
         verify(httpClient, times(1)).serverInfo();
         verifyNoMoreInteractions(httpClient);
     }
 
     public void testWrongServerVersion() throws Exception {
+        HttpClient httpClient = mock(HttpClient.class);
+        SqlVersion version = new SqlVersion((int)SqlVersion.V_7_7_0.major, SqlVersion.V_7_7_0.minor - 1, 0);
+        when(httpClient.serverInfo()).thenReturn(new MainResponse(randomAlphaOfLength(5), version.toString(),
+                ClusterName.DEFAULT.value(), UUIDs.randomBase64UUID()));
+        CliSession cliSession = new CliSession(httpClient);
+        Throwable throwable = expectThrows(ClientException.class, cliSession::checkConnection);
+        assertEquals("This version of the CLI is only compatible with Elasticsearch version " +
+            ClientVersion.CURRENT.majorMinorToString() + " or newer; attempting to connect to a server version " + version.toString(),
+            throwable.getMessage());
+        verify(httpClient, times(1)).serverInfo();
+        verifyNoMoreInteractions(httpClient);
+    }
+
+    public void testHigherServerVersion() throws Exception {
         HttpClient httpClient = mock(HttpClient.class);
         byte minor;
         byte major;
         if (randomBoolean()) {
-            minor = Version.CURRENT.minor;
-            major = (byte) (Version.CURRENT.major + 1);
+            minor = ClientVersion.CURRENT.minor;
+            major = (byte) (ClientVersion.CURRENT.major + 1);
         } else {
-            minor = (byte) (Version.CURRENT.minor + 1);
-            major = Version.CURRENT.major;
+            minor = (byte) (ClientVersion.CURRENT.minor + 1);
+            major = ClientVersion.CURRENT.major;
 
         }
         when(httpClient.serverInfo()).thenReturn(new MainResponse(randomAlphaOfLength(5),
-                org.elasticsearch.Version.fromString(major + "." + minor + ".23").toString(),
-                ClusterName.DEFAULT.value(), UUIDs.randomBase64UUID()));
+            SqlVersion.fromString(major + "." + minor + ".23").toString(),
+            ClusterName.DEFAULT.value(), UUIDs.randomBase64UUID()));
         CliSession cliSession = new CliSession(httpClient);
-        expectThrows(ClientException.class, cliSession::checkConnection);
+        cliSession.checkConnection();
         verify(httpClient, times(1)).serverInfo();
         verifyNoMoreInteractions(httpClient);
     }

x-pack/plugin/sql/sql-cli/src/test/java/org/elasticsearch/xpack/sql/cli/VersionTests.java

@@ -5,17 +5,17 @@
  */
 package org.elasticsearch.xpack.sql.cli;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.test.ESTestCase;
-import org.elasticsearch.xpack.sql.client.Version;
+import org.elasticsearch.xpack.sql.client.ClientVersion;
 
 public class VersionTests extends ESTestCase {
     public void testVersionIsCurrent() {
         /* This test will only work properly in gradle because in gradle we run the tests
          * using the jar. */
-        assertNotNull(Version.CURRENT.hash);
-        assertEquals(org.elasticsearch.Version.CURRENT.major, Version.CURRENT.major);
-        assertEquals(org.elasticsearch.Version.CURRENT.minor, Version.CURRENT.minor);
-        assertEquals(org.elasticsearch.Version.CURRENT.revision, Version.CURRENT.revision);
+        assertEquals(Version.CURRENT.major, ClientVersion.CURRENT.major);
+        assertEquals(Version.CURRENT.minor, ClientVersion.CURRENT.minor);
+        assertEquals(Version.CURRENT.revision, ClientVersion.CURRENT.revision);
     }
 
 }

x-pack/plugin/sql/sql-cli/src/test/java/org/elasticsearch/xpack/sql/cli/command/BuiltinCommandTests.java

@@ -7,8 +7,8 @@ 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.ClientVersion;
 import org.elasticsearch.xpack.sql.client.HttpClient;
-import org.elasticsearch.xpack.sql.client.Version;
 
 import static org.hamcrest.Matchers.containsString;
 import static org.mockito.Mockito.mock;
@@ -94,7 +94,7 @@ public class BuiltinCommandTests extends ESTestCase {
         testTerminal.print("not clean");
         assertTrue(new PrintLogoCommand().handle(testTerminal, cliSession, "logo"));
         assertThat(testTerminal.toString(), containsString("SQL"));
-        assertThat(testTerminal.toString(), containsString(Version.CURRENT.version));
+        assertThat(testTerminal.toString(), containsString(ClientVersion.CURRENT.version));
         verifyNoMoreInteractions(httpClient);
     }
 

x-pack/plugin/sql/sql-client/src/main/java/org/elasticsearch/xpack/sql/client/Version.java → x-pack/plugin/sql/sql-client/src/main/java/org/elasticsearch/xpack/sql/client/ClientVersion.java

@@ -5,6 +5,8 @@
  */
 package org.elasticsearch.xpack.sql.client;
 
+import org.elasticsearch.xpack.sql.proto.SqlVersion;
+
 import java.io.IOException;
 import java.net.URL;
 import java.net.URLConnection;
@@ -16,45 +18,25 @@ import java.util.Set;
 import java.util.jar.JarInputStream;
 import java.util.jar.Manifest;
 
-public class Version {
-
-    public static final Version CURRENT;
-    public final String version;
-    public final String hash;
-    public final byte major;
-    public final byte minor;
-    public final byte revision;
-
-    private Version(String version, String hash, byte... parts) {
-        this.version = version;
-        this.hash = hash;
-        this.major = parts[0];
-        this.minor = parts[1];
-        this.revision = parts[2];
-    }
-
-    public static Version fromString(String version) {
-        return new Version(version, "Unknown", from(version));
-    }
+/**
+ * Clients-specific version utility class.
+ * <p>
+ *     The class provides the SQL clients the version identifying the release they're are part of. The version is read from the
+ *     encompassing JAR file (Elasticsearch-specific attribute in the manifest).
+ *     The class is also a provider for the implemented JDBC standard.
+ * </p>
+ */
+public class ClientVersion {
 
-    static byte[] from(String ver) {
-        String[] parts = ver.split("[.-]");
-        // Allow for optional snapshot and qualifier
-        if (parts.length < 3 || parts.length > 5) {
-            throw new IllegalArgumentException("Invalid version " + ver);
-        }
-        else {
-            return new byte[] { Byte.parseByte(parts[0]), Byte.parseByte(parts[1]), Byte.parseByte(parts[2]) };
-        }
-    }
+    public static final SqlVersion CURRENT;
 
     static {
         // check classpath
-        String target = Version.class.getName().replace(".", "/").concat(".class");
+        String target = ClientVersion.class.getName().replace(".", "/").concat(".class");
         Enumeration<URL> res;
 
         try {
-            res = Version.class.getClassLoader().getResources(target);
+            res = ClientVersion.class.getClassLoader().getResources(target);
         } catch (IOException ex) {
             throw new IllegalArgumentException("Cannot detect Elasticsearch JDBC jar; it typically indicates a deployment issue...");
         }
@@ -85,41 +67,35 @@ public class Version {
         }
 
         // This is similar to how Elasticsearch's Build class digs up its build information.
-        // Since version info is not critical, the parsing is lenient
-        URL url = Version.class.getProtectionDomain().getCodeSource().getLocation();
+        URL url = SqlVersion.class.getProtectionDomain().getCodeSource().getLocation();
         CURRENT = extractVersion(url);
     }
 
-    static Version extractVersion(URL url) {
+    static SqlVersion extractVersion(URL url) {
         String urlStr = url.toString();
-        byte maj = 0, min = 0, rev = 0;
-        String ver = "Unknown";
-        String hash = ver;
-        
         if (urlStr.endsWith(".jar") || urlStr.endsWith(".jar!/")) {
             try {
                 URLConnection conn = url.openConnection();
                 conn.setUseCaches(false);
-                
+
                 try (JarInputStream jar = new JarInputStream(conn.getInputStream())) {
                     Manifest manifest = jar.getManifest();
-                    hash = manifest.getMainAttributes().getValue("Change");
-                    ver = manifest.getMainAttributes().getValue("X-Compile-Elasticsearch-Version");
-                    byte[] vers = from(ver);
-                    maj = vers[0];
-                    min = vers[1];
-                    rev = vers[2];
+                    String version = manifest.getMainAttributes().getValue("X-Compile-Elasticsearch-Version");
+                    return SqlVersion.fromString(version);
                 }
             } catch (Exception ex) {
                 throw new IllegalArgumentException("Detected Elasticsearch JDBC jar but cannot retrieve its version", ex);
             }
         }
-        return new Version(ver, hash, maj, min, rev);
+        return new SqlVersion(0, 0, 0);
     }
 
-    @Override
-    public String toString() {
-        return "v" + version + " [" + hash + "]";
+    // This function helps ensure that a client won't attempt to communicate to a server with less features than its own. Since this check
+    // is part of the client's start-up check that might not involve an actual SQL API request, the client has to do a bare version check
+    // as well.
+    public static boolean isServerCompatible(SqlVersion server) {
+        // Starting with this version, the compatibility logic moved from the client to the server.
+        return SqlVersion.hasVersionCompatibility(server);
     }
 
     public static int jdbcMajorVersion() {

x-pack/plugin/sql/sql-client/src/main/java/org/elasticsearch/xpack/sql/client/ConnectionConfiguration.java

@@ -57,8 +57,6 @@ public class ConnectionConfiguration {
 
     public static final String PAGE_SIZE = "page.size";
     private static final String PAGE_SIZE_DEFAULT = "1000";
-    
-    public static final String CLIENT_ID = "client_id";
 
     // Auth
 

x-pack/plugin/sql/sql-client/src/main/java/org/elasticsearch/xpack/sql/client/HttpClient.java

@@ -70,7 +70,7 @@ public class HttpClient {
                 null,
                 Boolean.FALSE,
                 null,
-                new RequestInfo(Mode.CLI),
+                new RequestInfo(Mode.CLI, ClientVersion.CURRENT),
                 false,
                 false,
                 cfg.binaryCommunication());

x-pack/plugin/sql/sql-client/src/test/java/org/elasticsearch/xpack/sql/client/HttpClientRequestTests.java

@@ -158,7 +158,7 @@ public class HttpClientRequestTests extends ESTestCase {
                 null,
                 randomBoolean(),
                 randomAlphaOfLength(128),
-                new RequestInfo(mode),
+                new RequestInfo(mode, ClientVersion.CURRENT),
                 randomBoolean(),
                 randomBoolean(),
                 isBinary);

x-pack/plugin/sql/sql-client/src/test/java/org/elasticsearch/xpack/sql/client/VersionTests.java

@@ -6,6 +6,7 @@
 package org.elasticsearch.xpack.sql.client;
 
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.sql.proto.SqlVersion;
 
 import java.io.BufferedInputStream;
 import java.io.IOException;
@@ -19,45 +20,21 @@ import java.util.jar.JarOutputStream;
 import java.util.jar.Manifest;
 
 public class VersionTests extends ESTestCase {
-    public void test70Version() {
-        byte[] ver = Version.from("7.0.0-alpha");
-        assertEquals(7, ver[0]);
-        assertEquals(0, ver[1]);
-        assertEquals(0, ver[2]);
-    }
-
-    public void test712Version() {
-        byte[] ver = Version.from("7.1.2");
-        assertEquals(7, ver[0]);
-        assertEquals(1, ver[1]);
-        assertEquals(2, ver[2]);
-    }
 
     public void testCurrent() {
-        Version ver = Version.fromString(org.elasticsearch.Version.CURRENT.toString());
-        assertEquals(org.elasticsearch.Version.CURRENT.major, ver.major);
-        assertEquals(org.elasticsearch.Version.CURRENT.minor, ver.minor);
-        assertEquals(org.elasticsearch.Version.CURRENT.revision, ver.revision);
-    }
-
-    public void testFromString() {
-        Version ver = Version.fromString("1.2.3");
-        assertEquals(1, ver.major);
-        assertEquals(2, ver.minor);
-        assertEquals(3, ver.revision);
-        assertEquals("Unknown", ver.hash);
-        assertEquals("1.2.3", ver.version);
+        SqlVersion ver = SqlVersion.fromString(org.elasticsearch.Version.CURRENT.toString());
+        assertEquals(ver, ClientVersion.CURRENT);
     }
 
     public void testInvalidVersion() {
-        IllegalArgumentException err = expectThrows(IllegalArgumentException.class, () -> Version.from("7.1"));
-        assertEquals("Invalid version 7.1", err.getMessage());
+        IllegalArgumentException err = expectThrows(IllegalArgumentException.class, () -> SqlVersion.fromString("7.1"));
+        assertEquals("Invalid version format [7.1]", err.getMessage());
     }
-    
+
     public void testVersionFromJarInJar() throws IOException {
         final String JDBC_JAR_NAME = "es-sql-jdbc.jar";
         final String JAR_PATH_SEPARATOR = "!/";
-        
+
         Path dir = createTempDir();
         Path jarPath = dir.resolve("uberjar.jar");          // simulated uberjar containing the jdbc driver
         Path innerJarPath = dir.resolve(JDBC_JAR_NAME); // simulated ES JDBC driver file
@@ -88,12 +65,11 @@ public class VersionTests extends ESTestCase {
         }
         
         URL jarInJar = new URL("jar:" + jarPath.toUri().toURL().toString() + JAR_PATH_SEPARATOR + JDBC_JAR_NAME + JAR_PATH_SEPARATOR);
-        
-        Version version = Version.extractVersion(jarInJar);
+
+        SqlVersion version = ClientVersion.extractVersion(jarInJar);
         assertEquals(1, version.major);
         assertEquals(2, version.minor);
         assertEquals(3, version.revision);
-        assertEquals("abc", version.hash);
         assertEquals("1.2.3", version.version);
     }
 }

x-pack/plugin/sql/sql-proto/src/main/java/org/elasticsearch/xpack/sql/proto/AbstractSqlRequest.java

@@ -34,6 +34,10 @@ public abstract class AbstractSqlRequest implements ToXContentFragment {
         return requestInfo.clientId();
     }
 
+    public SqlVersion version() {
+        return requestInfo.version();
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;

x-pack/plugin/sql/sql-proto/src/main/java/org/elasticsearch/xpack/sql/proto/Mode.java

@@ -33,4 +33,8 @@ public enum Mode {
     public static boolean isDriver(Mode mode) {
         return mode == JDBC || mode == ODBC;
     }
+
+    public static boolean isDedicatedClient(Mode mode) {
+        return mode == JDBC || mode == ODBC || mode == CLI;
+    }
 }

x-pack/plugin/sql/sql-proto/src/main/java/org/elasticsearch/xpack/sql/proto/RequestInfo.java

@@ -35,14 +35,25 @@ public class RequestInfo {
     
     private Mode mode;
     private String clientId;
-    
+    private SqlVersion version;
+
     public RequestInfo(Mode mode) {
-        this(mode, null);
+        this(mode, null, null);
     }
     
     public RequestInfo(Mode mode, String clientId) {
+        this(mode, clientId, null);
+    }
+
+    public RequestInfo(Mode mode, String clientId, String version) {
         mode(mode);
         clientId(clientId);
+        version(version);
+    }
+
+    public RequestInfo(Mode mode, SqlVersion version) {
+        mode(mode);
+        this.version = version;
     }
     
     public Mode mode() {
@@ -67,6 +78,14 @@ public class RequestInfo {
         this.clientId = clientId;
     }
 
+    public void version(String clientVersion) {
+        this.version = SqlVersion.fromString(clientVersion);
+    }
+
+    public SqlVersion version() {
+        return version;
+    }
+
     @Override
     public int hashCode() {
         return Objects.hash(mode, clientId);

x-pack/plugin/sql/sql-proto/src/main/java/org/elasticsearch/xpack/sql/proto/SqlClearCursorRequest.java

@@ -47,6 +47,9 @@ public class SqlClearCursorRequest extends AbstractSqlRequest {
         if (clientId() != null) {
             builder.field("client_id", clientId());
         }
+        if (version() != null) {
+            builder.field("version", version().toString());
+        }
         return builder;
     }
 }

x-pack/plugin/sql/sql-proto/src/main/java/org/elasticsearch/xpack/sql/proto/SqlQueryRequest.java

@@ -180,6 +180,9 @@ public class SqlQueryRequest extends AbstractSqlRequest {
         if (clientId() != null) {
             builder.field("client_id", clientId());
         }
+        if (version() != null) {
+            builder.field("version", version().toString());
+        }
         if (this.params != null && this.params.isEmpty() == false) {
             builder.startArray("params");
             for (SqlTypedParamValue val : this.params) {

x-pack/plugin/sql/sql-proto/src/main/java/org/elasticsearch/xpack/sql/proto/SqlVersion.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.proto;
+
+import java.security.InvalidParameterException;
+
+/**
+ * Elasticsearch's version modeler for the SQL plugin.
+ * <p>
+ *     The class models the version that the Elasticsearch server and the clients use to identify their release by. It is similar to
+ *     server's <strong>Version</strong> class (which is unavailable in this package), specific to the SQL plugin and its clients, and an
+ *     aid to establish the compatibility between them.
+ * </p>
+ */
+public class SqlVersion implements Comparable<SqlVersion>{
+
+    public final int id;
+    public final String version; // originally provided String representation
+    public final byte major;
+    public final byte minor;
+    public final byte revision;
+
+    public static final int REVISION_MULTIPLIER = 100;
+    public static final int MINOR_MULTIPLIER = REVISION_MULTIPLIER * REVISION_MULTIPLIER;
+    public static final int MAJOR_MULTIPLIER = REVISION_MULTIPLIER * MINOR_MULTIPLIER;
+
+    public static final SqlVersion V_7_7_0 = new SqlVersion(7, 7, 0);
+
+    public SqlVersion(byte major, byte minor, byte revision) {
+        this(toString(major, minor, revision), major, minor, revision);
+    }
+
+    public SqlVersion(Integer major, Integer minor, Integer revision) {
+        this(major.byteValue(), minor.byteValue(), revision.byteValue());
+        if (major > Byte.MAX_VALUE || minor > Byte.MAX_VALUE || revision > Byte.MAX_VALUE) {
+            throw new InvalidParameterException("Invalid version initialisers [" + major + ", " + minor + ", " + revision + "]");
+        }
+    }
+
+    protected SqlVersion(String version, byte... parts) {
+        this.version = version;
+
+        assert parts.length >= 3 : "Version must be initialized with all Major.Minor.Revision components";
+        this.major = parts[0];
+        this.minor = parts[1];
+        this.revision = parts[2];
+
+        if ((major | minor | revision) < 0 || minor >= REVISION_MULTIPLIER || revision >= REVISION_MULTIPLIER) {
+            throw new InvalidParameterException("Invalid version initialisers [" + major + ", " + minor + ", " + revision + "]");
+        }
+
+        id = Integer.valueOf(major) * MAJOR_MULTIPLIER
+            + Integer.valueOf(minor) * MINOR_MULTIPLIER
+            + Integer.valueOf(revision) * REVISION_MULTIPLIER;
+    }
+
+    public static SqlVersion fromString(String version) {
+        if (version == null || version.isEmpty()) {
+            return null;
+        }
+        return new SqlVersion(version, from(version));
+    }
+
+    protected static byte[] from(String ver) {
+        String[] parts = ver.split("[.-]");
+        // Allow for optional snapshot and qualifier (Major.Minor.Revision-Qualifier-SNAPSHOT)
+        if (parts.length >= 3 && parts.length <= 5) {
+            try {
+                return new byte[] { Byte.parseByte(parts[0]), Byte.parseByte(parts[1]), Byte.parseByte(parts[2]) };
+            } catch (NumberFormatException nfe) {
+                throw new IllegalArgumentException("Invalid version format [" + ver + "]: " + nfe.getMessage());
+            }
+        } else {
+            throw new IllegalArgumentException("Invalid version format [" + ver + "]");
+        }
+
+    }
+
+    private static String toString(byte... parts) {
+        assert parts.length >= 1 : "Version must contain at least a Major component";
+        String ver = String.valueOf(parts[0]);
+        for (int i = 1; i < parts.length; i ++) {
+            ver += "." + parts[i];
+        }
+        return ver;
+    }
+
+    @Override
+    public String toString() {
+        return toString(major, minor, revision);
+    }
+
+    public String majorMinorToString() {
+        return toString(major, minor);
+    }
+
+    @Override
+    public int hashCode() {
+        return id;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null) {
+            return false;
+        }
+        if (o.getClass() == getClass()) {
+            return ((SqlVersion) o).id == id;
+        }
+        if (o.getClass() == String.class) {
+            try {
+                SqlVersion v = fromString((String) o);
+                return this.equals(v);
+            } catch (IllegalArgumentException e) {
+                return false;
+            }
+        }
+        return false;
+    }
+
+    @Override
+    public int compareTo(SqlVersion o) {
+        return id - o.id;
+    }
+
+    public static int majorMinorId(SqlVersion v) {
+        return v.major * MAJOR_MULTIPLIER + v.minor * MINOR_MULTIPLIER;
+    }
+
+    public int compareToMajorMinor(SqlVersion o) {
+        return majorMinorId(this) - majorMinorId(o);
+    }
+
+    public static boolean hasVersionCompatibility(SqlVersion version) {
+        return version.compareTo(V_7_7_0) >= 0;
+    }
+}

x-pack/plugin/sql/sql-proto/src/test/java/org/elasticsearch/xpack/sql/proto/SqlVersionTests.java

@@ -0,0 +1,63 @@
+/*
+ * 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.proto;
+
+import org.elasticsearch.test.ESTestCase;
+
+public class SqlVersionTests extends ESTestCase {
+    public void test123FromString() {
+        SqlVersion ver = SqlVersion.fromString("1.2.3");
+        assertEquals(1, ver.major);
+        assertEquals(2, ver.minor);
+        assertEquals(3, ver.revision);
+        assertEquals(1 * SqlVersion.MAJOR_MULTIPLIER + 2 * SqlVersion.MINOR_MULTIPLIER + 3 * SqlVersion.REVISION_MULTIPLIER, ver.id);
+        assertEquals("1.2.3", ver.version);
+    }
+
+    public void test123AlphaFromString() {
+        SqlVersion ver = SqlVersion.fromString("1.2.3-Alpha");
+        assertEquals(1, ver.major);
+        assertEquals(2, ver.minor);
+        assertEquals(3, ver.revision);
+        assertEquals(1 * SqlVersion.MAJOR_MULTIPLIER + 2 * SqlVersion.MINOR_MULTIPLIER + 3 * SqlVersion.REVISION_MULTIPLIER, ver.id);
+        assertEquals("1.2.3-Alpha", ver.version);
+    }
+
+    public void test123AlphaSnapshotFromString() {
+        SqlVersion ver = SqlVersion.fromString("1.2.3-Alpha-SNAPSHOT");
+        assertEquals(1, ver.major);
+        assertEquals(2, ver.minor);
+        assertEquals(3, ver.revision);
+        assertEquals(1 * SqlVersion.MAJOR_MULTIPLIER + 2 * SqlVersion.MINOR_MULTIPLIER + 3 * SqlVersion.REVISION_MULTIPLIER, ver.id);
+        assertEquals("1.2.3-Alpha-SNAPSHOT", ver.version);
+    }
+
+    public void testVersionsEqual() {
+        SqlVersion ver1 = SqlVersion.fromString("1.2.3");
+        SqlVersion ver2 = SqlVersion.fromString("1.2.3");
+        assertEquals(ver1, ver2);
+    }
+
+    public void testVersionsAndStringEqual() {
+        SqlVersion ver1 = SqlVersion.fromString("1.2.3");
+        String ver2 = "1.2.3";
+        assertEquals(ver1, ver2);
+    }
+
+    public void testVersionsAndStringNotEqual() {
+        SqlVersion ver1 = SqlVersion.fromString("1.2.3");
+        String ver2 = "1.2.4";
+        assertNotEquals(ver1, ver2);
+    }
+
+    public void testVersionsAndInvalidStringNotEqual() {
+        SqlVersion ver1 = SqlVersion.fromString("1.2.3");
+        String ver2 = "invalid";
+        assertNotEquals(ver1, ver2);
+    }
+
+}

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

@@ -60,7 +60,7 @@ public class RestSqlQueryAction extends BaseRestHandler {
          */
         String accept = null;
 
-        if ((Mode.isDriver(sqlRequest.requestInfo().mode()) || sqlRequest.requestInfo().mode() == Mode.CLI)
+        if (Mode.isDedicatedClient(sqlRequest.requestInfo().mode())
                 && (sqlRequest.binaryCommunication() == null || sqlRequest.binaryCommunication())) {
             // enforce CBOR response for drivers and CLI (unless instructed differently through the config param)
             accept = XContentType.CBOR.name();

x-pack/plugin/sql/src/test/java/org/elasticsearch/xpack/sql/action/SqlActionIT.java

@@ -5,12 +5,14 @@
  */
 package org.elasticsearch.xpack.sql.action;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.support.WriteRequest;
 import org.elasticsearch.xpack.sql.proto.ColumnInfo;
 import org.elasticsearch.xpack.sql.proto.Mode;
 
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertRequestBuilderThrows;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasSize;
 
@@ -28,7 +30,7 @@ public class SqlActionIT extends AbstractSqlIntegTestCase {
         boolean dataBeforeCount = randomBoolean();
         String columns = dataBeforeCount ? "data, count" : "count, data";
         SqlQueryResponse response = new SqlQueryRequestBuilder(client(), SqlQueryAction.INSTANCE)
-                .query("SELECT " + columns + " FROM test ORDER BY count").mode(Mode.JDBC).get();
+                .query("SELECT " + columns + " FROM test ORDER BY count").mode(Mode.JDBC).version(Version.CURRENT.toString()).get();
         assertThat(response.size(), equalTo(2L));
         assertThat(response.columns(), hasSize(2));
         int dataIndex = dataBeforeCount ? 0 : 1;
@@ -42,5 +44,18 @@ public class SqlActionIT extends AbstractSqlIntegTestCase {
         assertEquals("baz", response.rows().get(1).get(dataIndex));
         assertEquals(43L, response.rows().get(1).get(countIndex));
     }
+
+    public void testSqlActionCurrentVersion() {
+        SqlQueryResponse response = new SqlQueryRequestBuilder(client(), SqlQueryAction.INSTANCE)
+            .query("SELECT true").mode(randomFrom(Mode.CLI, Mode.JDBC)).version(Version.CURRENT.toString()).get();
+        assertThat(response.size(), equalTo(1L));
+        assertEquals(true, response.rows().get(0).get(0));
+    }
+
+    public void testSqlActionOutdatedVersion() {
+        SqlQueryRequestBuilder request = new SqlQueryRequestBuilder(client(), SqlQueryAction.INSTANCE)
+            .query("SELECT true").mode(randomFrom(Mode.CLI, Mode.JDBC)).version("1.2.3");
+        assertRequestBuilderThrows(request, org.elasticsearch.action.ActionRequestValidationException.class);
+    }
 }