JDBC: Fix stackoverflow on getObject and timestamp conversion (#31735)

StackOverflowError fix in JdbcResultSet getObject method.
Fix Timestamp conversion bug when getting the value of a time column.

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

@@ -344,7 +344,7 @@ class JdbcResultSet implements ResultSet, JdbcWrapper {
             throw new SQLException("type is null");
         }
 
-        return getObject(columnIndex, type);
+        return convert(columnIndex, type);
     }
 
     private <T> T convert(int columnIndex, Class<T> type) throws SQLException {

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

@@ -254,7 +254,7 @@ final class TypeConverter {
             case REAL:
                 return floatValue(v);  // Float might be represented as string for infinity and NaN values
             case TIMESTAMP:
-                return ((Number) v).longValue();
+                return new Timestamp(((Number) v).longValue());
             default:
                 throw new SQLException("Unexpected column type [" + columnType.getName() + "]");
 

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

@@ -14,6 +14,7 @@ import org.joda.time.DateTime;
 import org.joda.time.ReadableDateTime;
 
 import java.sql.JDBCType;
+import java.sql.Timestamp;
 
 import static org.hamcrest.Matchers.instanceOf;
 
@@ -41,8 +42,8 @@ public class TypeConverterTests extends ESTestCase {
 
     public void testTimestampAsNative() throws Exception {
         DateTime now = DateTime.now();
-        assertThat(convertAsNative(now, JDBCType.TIMESTAMP), instanceOf(Long.class));
+        assertThat(convertAsNative(now, JDBCType.TIMESTAMP), instanceOf(Timestamp.class));
-        assertEquals(now.getMillis(), convertAsNative(now, JDBCType.TIMESTAMP));
+        assertEquals(now.getMillis(), ((Timestamp) convertAsNative(now, JDBCType.TIMESTAMP)).getTime());
     }
 
     private Object convertAsNative(Object value, JDBCType type) throws Exception {

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

@@ -82,7 +82,11 @@ public abstract class JdbcIntegrationTestCase extends ESRestTestCase {
     }
 
     public static void index(String index, CheckedConsumer<XContentBuilder, IOException> body) throws IOException {
-        Request request = new Request("PUT", "/" + index + "/doc/1");
+        index(index, "1", body);
+    }
+    
+    public static void index(String index, String documentId, CheckedConsumer<XContentBuilder, IOException> body) throws IOException {
+        Request request = new Request("PUT", "/" + index + "/doc/" + documentId);
         request.addParameter("refresh", "true");
         XContentBuilder builder = JsonXContent.contentBuilder().startObject();
         body.accept(builder);

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

@@ -0,0 +1,82 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.qa.sql.jdbc;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Timestamp;
+import java.util.Date;
+
+public class ResultSetTestCase extends JdbcIntegrationTestCase {
+    public void testGettingTimestamp() throws Exception {
+        long randomMillis = randomLongBetween(0, System.currentTimeMillis());
+
+        index("library", "1", builder -> {
+            builder.field("name", "Don Quixote");
+            builder.field("page_count", 1072);
+            builder.timeField("release_date", new Date(randomMillis));
+            builder.timeField("republish_date", null);
+        });
+        index("library", "2", builder -> {
+            builder.field("name", "1984");
+            builder.field("page_count", 328);
+            builder.timeField("release_date", new Date(-649036800000L));
+            builder.timeField("republish_date", new Date(599616000000L));
+        });
+
+        try (Connection connection = esJdbc()) {
+            try (PreparedStatement statement = connection.prepareStatement("SELECT name, release_date, republish_date FROM library")) {
+                try (ResultSet results = statement.executeQuery()) {
+                    ResultSetMetaData resultSetMetaData = results.getMetaData();
+
+                    results.next();
+                    assertEquals(3, resultSetMetaData.getColumnCount());
+                    assertEquals(randomMillis, results.getTimestamp("release_date").getTime());
+                    assertEquals(randomMillis, results.getTimestamp(2).getTime());
+                    assertTrue(results.getObject(2) instanceof Timestamp);
+                    assertEquals(randomMillis, ((Timestamp) results.getObject("release_date")).getTime());
+                    
+                    assertNull(results.getTimestamp(3));
+                    assertNull(results.getObject("republish_date"));
+
+                    assertTrue(results.next());
+                    assertEquals(599616000000L, results.getTimestamp("republish_date").getTime());
+                    assertEquals(-649036800000L, ((Timestamp) results.getObject(2)).getTime());
+
+                    assertFalse(results.next());
+                }
+            }
+        }
+    }
+
+    /*
+     * Checks StackOverflowError fix for https://github.com/elastic/elasticsearch/pull/31735
+     */
+    public void testNoInfiniteRecursiveGetObjectCalls() throws SQLException, IOException {
+        index("library", "1", builder -> {
+            builder.field("name", "Don Quixote");
+            builder.field("page_count", 1072);
+        });
+        Connection conn = esJdbc();
+        PreparedStatement statement = conn.prepareStatement("SELECT * FROM library");
+        ResultSet results = statement.executeQuery();
+
+        try {
+            results.next();
+            results.getObject("name");
+            results.getObject("page_count");
+            results.getObject(1);
+            results.getObject(1, String.class);
+            results.getObject("page_count", Integer.class);
+        } catch (StackOverflowError soe) {
+            fail("Infinite recursive call on getObject() method");
+        }
+    }
+}