SQL: Fix issue with timezone for JDBC pagination

Previously, when the specified (or default) `fetchSize` led to
subsequent HTTP requests and the usage of cursors, those subsequent
cursor requests didn't use the `timezone` defined in the connection
properties. Even though the query is executed once (with the correct
timezone) the processing of the values returned by the `HitExtractors`
in the next pages was done using the default timezone `Z`. This could
lead to incorrect results.

Fix the issue by passing the initially configured timezone to each
subsequent cursor HTTP request.

Add a note in the docs to clarify that when the REST endpoint is used,
the user needs to keep passing the `time_zone` parameter to each
subsequent request, as there is no notion of a user session through in
the REST HTTP endpoint.

Relates to: #51258

docs/reference/sql/endpoints/rest.asciidoc

@@ -305,7 +305,12 @@ Which looks like:
 --------------------------------------------------
 // TESTRESPONSE[s/sDXF1ZXJ5QW5kRmV0Y2gBAAAAAAAAAAEWODRMaXBUaVlRN21iTlRyWHZWYUdrdw==:BAFmBmF1dGhvcgFmBG5hbWUBZgpwYWdlX2NvdW50AWYMcmVsZWFzZV9kYXRl9f\/\/\/w8=/$body.cursor/]
 
-Note that the `columns` object is only part of the first page.
+[NOTE]
+The `columns` object is only part of the first page.
+[NOTE]
+If in the initial HTTP request, that specifies the SQL query, there was a <<sql-rest-param-timezone, `time_zone`>>
+param passed, the same `time_zone` parameter should be passed to the subsequent `cursor` requests in order to keep
+applying the same `time_zone` (as in the initial HTTP request) the result data of each subsequent page.
 
 You've reached the last page when there is no `cursor` returned
 in the results. Like Elasticsearch's <<request-body-search-scroll,scroll>>,
@@ -509,6 +514,7 @@ s|Description
 |45s
 |The timeout before a pagination request fails.
 
+[[sql-rest-param-timezone]]
 |time_zone
 |`Z` (or `UTC`)
 |Time-zone in ISO 8601 used for executing the query on the server.
@@ -533,4 +539,4 @@ More information available https://docs.oracle.com/javase/8/docs/api/java/time/Z
 |===
 
 Do note that most parameters (outside the timeout and `columnar` ones) make sense only during the initial query - any follow-up pagination request only requires the `cursor` parameter as explained in the <<sql-pagination, pagination>> chapter.
-That's because the query has already been executed and the calls are simply about returning the found results - thus the parameters are simply ignored.
+That's because the query has already been executed and the calls are simply about returning the found results - thus the parameters are simply ignored.

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

@@ -75,7 +75,7 @@ Cursor query(String sql, List<SqlTypedParamValue> params, RequestMeta meta) thro
      * the scroll id to use to fetch the next page.
      */
     Tuple<String, List<List<Object>>> nextPage(String cursor, RequestMeta meta) throws SQLException {
-        SqlQueryRequest sqlRequest = new SqlQueryRequest(cursor, TimeValue.timeValueMillis(meta.timeoutInMs()),
+        SqlQueryRequest sqlRequest = new SqlQueryRequest(cursor, conCfg.zoneId(), TimeValue.timeValueMillis(meta.timeoutInMs()),
                 TimeValue.timeValueMillis(meta.queryTimeoutInMs()), new RequestInfo(Mode.JDBC), conCfg.binaryCommunication());
         SqlQueryResponse response = httpClient.query(sqlRequest);
         return new Tuple<>(response.cursor(), response.rows());

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

@@ -16,7 +16,12 @@
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.util.Properties;
 
+import static org.elasticsearch.xpack.sql.qa.jdbc.JdbcTestUtils.JDBC_TIMEZONE;
 import static org.elasticsearch.xpack.sql.qa.rest.RestSqlTestCase.assertNoSearchContexts;
 
 /**
@@ -121,6 +126,52 @@ public void testAggregation() throws SQLException {
             }
         }
     }
+
+    public void testWithDatetimeAndTimezoneParam() throws IOException, SQLException {
+        Request request = new Request("PUT", "/test_date_timezone");
+        XContentBuilder createIndex = JsonXContent.contentBuilder().startObject();
+        createIndex.startObject("mappings");
+        {
+            createIndex.startObject("properties");
+            {
+                createIndex.startObject("date").field("type", "date").field("format", "epoch_millis");
+                createIndex.endObject();
+            }
+            createIndex.endObject();
+        }
+        createIndex.endObject().endObject();
+        request.setJsonEntity(Strings.toString(createIndex));
+        client().performRequest(request);
+
+        request = new Request("PUT", "/test_date_timezone/_bulk");
+        request.addParameter("refresh", "true");
+        StringBuilder bulk = new StringBuilder();
+        long[] datetimes = new long[] { 1_000, 10_000, 100_000, 1_000_000, 10_000_000 };
+        for (int i = 0; i < datetimes.length; i++) {
+            bulk.append("{\"index\":{}}\n");
+            bulk.append("{\"date\":").append(datetimes[i]).append("}\n");
+        }
+        request.setJsonEntity(bulk.toString());
+        assertEquals(200, client().performRequest(request).getStatusLine().getStatusCode());
+
+        ZoneId zoneId = randomZone();
+        Properties connectionProperties = connectionProperties();
+        connectionProperties.put(JDBC_TIMEZONE, zoneId.toString());
+        try (Connection c = esJdbc(connectionProperties);
+             Statement s = c.createStatement()) {
+            s.setFetchSize(2);
+            try (ResultSet rs =
+                         s.executeQuery("SELECT DATE_PART('TZOFFSET', date) FROM test_date_timezone ORDER BY date")) {
+                for (int i = 0; i < datetimes.length; i++) {
+                    assertEquals(2, rs.getFetchSize());
+                    assertTrue("No more entries left at " + i, rs.next());
+                    assertEquals(ZonedDateTime.ofInstant(Instant.ofEpochMilli(datetimes[i]), zoneId).getOffset()
+                                    .getTotalSeconds()/ 60, rs.getInt(1));
+                }
+                assertFalse(rs.next());
+            }
+        }
+    }
 
     /**
      * Test for nested documents.
@@ -237,4 +288,4 @@ private void addPivotData() throws Exception {
         request.setJsonEntity(bulk.toString());
         assertEquals(200, client().performRequest(request).getStatusLine().getStatusCode());
     }
-}
+}

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

@@ -83,7 +83,7 @@ public SqlQueryResponse query(SqlQueryRequest sqlRequest) throws SQLException {
 
     public SqlQueryResponse nextPage(String cursor) throws SQLException {
         // method called only from CLI
-        SqlQueryRequest sqlRequest = new SqlQueryRequest(cursor, TimeValue.timeValueMillis(cfg.queryTimeout()),
+        SqlQueryRequest sqlRequest = new SqlQueryRequest(cursor, Protocol.TIME_ZONE, TimeValue.timeValueMillis(cfg.queryTimeout()),
                 TimeValue.timeValueMillis(cfg.pageTimeout()), new RequestInfo(Mode.CLI), cfg.binaryCommunication());
         return post(Protocol.SQL_QUERY_REST_ENDPOINT, sqlRequest, SqlQueryResponse::fromXContent);
     }

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

@@ -55,9 +55,9 @@ public SqlQueryRequest(String query, List<SqlTypedParamValue> params, ZoneId zon
         this.binaryCommunication = binaryCommunication;
     }
 
-    public SqlQueryRequest(String cursor, TimeValue requestTimeout, TimeValue pageTimeout, RequestInfo requestInfo,
+    public SqlQueryRequest(String cursor, ZoneId zoneId, TimeValue requestTimeout, TimeValue pageTimeout, RequestInfo requestInfo,
                            boolean binaryCommunication) {
-        this("", Collections.emptyList(), Protocol.TIME_ZONE, Protocol.FETCH_SIZE, requestTimeout, pageTimeout,
+        this("", Collections.emptyList(), zoneId, Protocol.FETCH_SIZE, requestTimeout, pageTimeout,
                 null, false, cursor, requestInfo, Protocol.FIELD_MULTI_VALUE_LENIENCY, Protocol.INDEX_INCLUDE_FROZEN, binaryCommunication);
     }
 
@@ -220,4 +220,4 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         }
         return builder;
     }
-}
+}