backwards compatible deserialization of SqlQueryResponse and fix SQL bwc tests (#78467)

* make deserialization of SqlQueryResponse backwards compatible and fix SQL bwc tests

* address comments

* reenable tests

Author: Lukas Wegmann

x-pack/plugin/sql/qa/mixed-node/build.gradle

@@ -21,8 +21,7 @@ testClusters.configureEach {
 tasks.named("integTest").configure{ enabled = false}
 
 // A bug (https://github.com/elastic/elasticsearch/issues/68439) limits us to perform tests with versions from 7.10.3 onwards
-
-BuildParams.bwcVersions.withWireCompatiple(v -> v.onOrAfter("7.10.0") &&
+BuildParams.bwcVersions.withWireCompatiple(v -> v.onOrAfter("7.10.3") &&
         v != VersionProperties.getElasticsearchVersion()) { bwcVersion, baseName ->
 
   def baseCluster = testClusters.register(baseName) {

x-pack/plugin/sql/qa/mixed-node/src/test/java/org/elasticsearch/xpack/sql/qa/mixed_node/SqlCompatIT.java

@@ -12,6 +12,9 @@
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.Response;
 import org.elasticsearch.client.RestClient;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.common.xcontent.json.JsonXContent;
 import org.elasticsearch.core.internal.io.IOUtils;
@@ -123,8 +126,8 @@ private List<Integer> runOrderByNullsLastQuery(RestClient queryClient) throws IO
         indexDocs.setJsonEntity(bulk.toString());
         client().performRequest(indexDocs);
 
-        Request query = new Request("GET", "_sql");
-        query.setJsonEntity("{\"query\":\"SELECT int FROM test GROUP BY 1 ORDER BY 1 NULLS LAST\"}");
+        Request query = new Request("POST", "_sql");
+        query.setJsonEntity(sqlQueryEntityWithOptionalMode("SELECT int FROM test GROUP BY 1 ORDER BY 1 NULLS LAST", bwcVersion));
         Response queryResponse = queryClient.performRequest(query);
 
         assertEquals(200, queryResponse.getStatusLine().getStatusCode());
@@ -135,4 +138,21 @@ private List<Integer> runOrderByNullsLastQuery(RestClient queryClient) throws IO
         return rows.stream().map(row -> (Integer) row.get(0)).collect(Collectors.toList());
     }
 
+    public static String sqlQueryEntityWithOptionalMode(String query, Version bwcVersion) throws IOException {
+        XContentBuilder json = XContentFactory.jsonBuilder().startObject();
+        json.field("query", query);
+        if (bwcVersion.before(Version.V_7_12_0)) {
+            // a bug previous to 7.12 caused a NullPointerException when accessing displaySize in ColumnInfo. The bug has been addressed in
+            // https://github.com/elastic/elasticsearch/pull/68802/files
+            // #diff-2faa4e2df98a4636300a19d9d890a1bd7174e9b20dd3a8589d2c78a3d9e5cbc0L110
+            // as a workaround, use JDBC (driver) mode in versions prior to 7.12
+            json.field("mode", "jdbc");
+            json.field("binary_format", false);
+            json.field("version", bwcVersion.toString());
+        }
+        json.endObject();
+
+        return Strings.toString(json);
+    }
+
 }

x-pack/plugin/sql/qa/mixed-node/src/test/java/org/elasticsearch/xpack/sql/qa/mixed_node/SqlSearchIT.java

@@ -38,7 +38,6 @@
 import static java.util.Collections.unmodifiableMap;
 import static org.elasticsearch.xpack.ql.TestUtils.buildNodeAndVersions;
 import static org.elasticsearch.xpack.ql.TestUtils.readResource;
-import static org.elasticsearch.xpack.ql.execution.search.QlSourceBuilder.SWITCH_TO_FIELDS_API_VERSION;
 
 public class SqlSearchIT extends ESRestTestCase {
 
@@ -56,9 +55,7 @@ public class SqlSearchIT extends ESRestTestCase {
     private static List<TestNode> newNodes;
     private static List<TestNode> bwcNodes;
     private static Version bwcVersion;
-    private static Version newVersion;
     private static boolean isBwcNodeBeforeFieldsApiInQL;
-    private static boolean isBwcNodeBeforeFieldsApiInES;
 
     @Before
     public void createIndex() throws IOException {
@@ -68,9 +65,7 @@ public void createIndex() throws IOException {
         newNodes = new ArrayList<>(nodes.getNewNodes());
         bwcNodes = new ArrayList<>(nodes.getBWCNodes());
         bwcVersion = nodes.getBWCNodes().get(0).getVersion();
-        newVersion = nodes.getNewNodes().get(0).getVersion();
         isBwcNodeBeforeFieldsApiInQL = bwcVersion.before(FIELDS_API_QL_INTRODUCTION);
-        isBwcNodeBeforeFieldsApiInES = bwcVersion.before(SWITCH_TO_FIELDS_API_VERSION);
 
         String mappings = readResource(SqlSearchIT.class.getResourceAsStream("/all_field_types.json"));
         createIndex(
@@ -142,7 +137,7 @@ public void testAllTypesWithRequestToUpgradedNodes() throws Exception {
             (builder, fieldValues) -> {
                 Float randomFloat = randomFloat();
                 builder.append(",");
-                if (isBwcNodeBeforeFieldsApiInQL && isBwcNodeBeforeFieldsApiInES) {
+                if (isBwcNodeBeforeFieldsApiInQL) {
                     builder.append("\"geo_point_field\":{\"lat\":\"37.386483\", \"lon\":\"-122.083843\"},");
                     fieldValues.put("geo_point_field", "POINT (-122.08384302444756 37.38648299127817)");
                     builder.append("\"float_field\":" + randomFloat + ",");
@@ -256,20 +251,38 @@ private void assertAllTypesWithNodes(Map<String, Object> expectedResponse, List<
         ) {
             @SuppressWarnings("unchecked")
             List<Map<String, Object>> columns = (List<Map<String, Object>>) expectedResponse.get("columns");
+
             String intervalYearMonth = "INTERVAL '150' YEAR AS interval_year, ";
             String intervalDayTime = "INTERVAL '163' MINUTE AS interval_minute, ";
-
             // get all fields names from the expected response built earlier, skipping the intervals as they execute locally
             // and not taken from the index itself
-            String fieldsList = columns.stream().map(m -> (String) m.get("name")).filter(str -> str.startsWith("interval") == false)
-                .collect(Collectors.toList()).stream().collect(Collectors.joining(", "));
+            String fieldsList = columns.stream()
+                .map(m -> (String) m.get("name"))
+                .filter(str -> str.startsWith("interval") == false)
+                .collect(Collectors.toList())
+                .stream()
+                .collect(Collectors.joining(", "));
             String query = "SELECT " + intervalYearMonth + intervalDayTime + fieldsList + " FROM " + index + " ORDER BY id";
+
             Request request = new Request("POST", "_sql");
-            request.setJsonEntity("{\"query\":\"" + query + "\"}");
-            assertBusy(() -> { assertResponse(expectedResponse, runSql(client, request)); });
+            request.setJsonEntity(SqlCompatIT.sqlQueryEntityWithOptionalMode(query, bwcVersion));
+            assertBusy(() -> {
+                assertResponse(expectedResponse, dropDisplaySizes(runSql(client, request)));
+            });
         }
     }
 
+    private Map<String, Object> dropDisplaySizes(Map<String, Object> response) {
+        // in JDBC mode, display_size will be part of the response, so remove it because it's not part of the expected response
+        @SuppressWarnings("unchecked")
+        List<Map<String, Object>> columns = (List<Map<String, Object>>) response.get("columns");
+        List<Map<String, Object>> columnsWithoutDisplaySizes = columns.stream()
+            .peek(column -> column.remove("display_size"))
+            .collect(Collectors.toList());
+        response.put("columns", columnsWithoutDisplaySizes);
+        return response;
+    }
+
     private void assertResponse(Map<String, Object> expected, Map<String, Object> actual) {
         if (false == expected.equals(actual)) {
             NotEqualMessageBuilder message = new NotEqualMessageBuilder();

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

@@ -6,26 +6,27 @@
  */
 package org.elasticsearch.xpack.sql.action;
 
-import java.io.IOException;
-import java.time.ZonedDateTime;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Objects;
-
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionResponse;
-import org.elasticsearch.core.Nullable;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.xcontent.ToXContentObject;
 import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.core.Nullable;
 import org.elasticsearch.xpack.ql.async.QlStatusResponse;
 import org.elasticsearch.xpack.sql.proto.ColumnInfo;
 import org.elasticsearch.xpack.sql.proto.Mode;
 import org.elasticsearch.xpack.sql.proto.Protocol;
 import org.elasticsearch.xpack.sql.proto.SqlVersion;
 import org.elasticsearch.xpack.sql.proto.StringUtils;
 
+import java.io.IOException;
+import java.time.ZonedDateTime;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+
 import static java.util.Collections.unmodifiableList;
 import static org.elasticsearch.Version.CURRENT;
 import static org.elasticsearch.xpack.sql.action.AbstractSqlQueryRequest.CURSOR;
@@ -81,10 +82,16 @@ public SqlQueryResponse(StreamInput in) throws IOException {
             }
         }
         this.rows = unmodifiableList(rows);
-        columnar = in.readBoolean();
-        asyncExecutionId = in.readOptionalString();
-        isPartial = in.readBoolean();
-        isRunning = in.readBoolean();
+        if (in.getVersion().onOrAfter(Version.V_7_14_0)) {
+            columnar = in.readBoolean();
+            asyncExecutionId = in.readOptionalString();
+            isPartial = in.readBoolean();
+            isRunning = in.readBoolean();
+        } else {
+            asyncExecutionId = null;
+            isPartial = false;
+            isRunning = false;
+        }
     }
 
     public SqlQueryResponse(