SQL: Fix deserialization issue of TimeProcessor

TimeProcessor didn't implement `getWriteableName()` so the one from
the parent was used which returned the `NAME` of the parent. This
caused `TimeProcessor` objects to be deserialised into
DateTimeProcessor.

Moreover, added a restriction to run the TIME related integration tests
only in UTC timezone.

Fixes: #40717

Author: matriv

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

@@ -39,6 +39,7 @@
 import static org.elasticsearch.xpack.sql.jdbc.EsType.DATETIME;
 import static org.elasticsearch.xpack.sql.jdbc.EsType.TIME;
 import static org.elasticsearch.xpack.sql.jdbc.JdbcDateUtils.asDateTimeField;
+import static org.elasticsearch.xpack.sql.jdbc.JdbcDateUtils.timeAsTime;
 
 /**
  * Conversion utilities for conversion of JDBC types to Java type and back
@@ -220,7 +221,7 @@ static Object convert(Object v, EsType columnType, String typeString) throws SQL
             case DATE:
                 return asDateTimeField(v, JdbcDateUtils::asDate, Date::new);
             case TIME:
-                return asDateTimeField(v, JdbcDateUtils::asTime, Time::new);
+                return timeAsTime(v.toString());
             case DATETIME:
                 return asDateTimeField(v, JdbcDateUtils::asTimestamp, Timestamp::new);
             case INTERVAL_YEAR:

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

@@ -40,12 +40,14 @@ public CsvSpecTestCase(String fileName, String groupName, String testName, Integ
 
     @Override
     protected final void doTest() throws Throwable {
-        try (Connection csv = csvConnection(testCase); Connection es = esJdbc()) {
-
-            // pass the testName as table for debugging purposes (in case the underlying reader is missing)
-            ResultSet expected = executeCsvQuery(csv, testName);
-            ResultSet elasticResults = executeJdbcQuery(es, testCase.query);
-            assertResults(expected, elasticResults);
+        if ("time".equals(groupName)) { // Run the time tests always in UTC
+            try (Connection csv = csvConnection(testCase); Connection es = esJdbc(connectionProperties())) {
+                executeAndAssert(csv, es);
+            }
+        } else {
+            try (Connection csv = csvConnection(testCase); Connection es = esJdbc()) {
+                executeAndAssert(csv, es);
+            }
         }
     }
 
@@ -54,4 +56,11 @@ protected void assertResults(ResultSet expected, ResultSet elastic) throws SQLEx
         Logger log = logEsResultSet() ? logger : null;
         JdbcAssert.assertResultSets(expected, elastic, log, false, true);
     }
+
+    private void executeAndAssert(Connection csv, Connection es) throws SQLException {
+        // pass the testName as table for debugging purposes (in case the underlying reader is missing)
+        ResultSet expected = executeCsvQuery(csv, testName);
+        ResultSet elasticResults = executeJdbcQuery(es, testCase.query);
+        assertResults(expected, elasticResults);
+    }
 }

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

@@ -27,6 +27,7 @@
 import java.util.Properties;
 import java.util.Set;
 
+import static org.elasticsearch.xpack.sql.qa.jdbc.JdbcTestUtils.JDBC_TIMEZONE;
 import static org.elasticsearch.xpack.sql.qa.rest.RestSqlTestCase.assertNoSearchContexts;
 
 public abstract class JdbcIntegrationTestCase extends ESRestTestCase {
@@ -137,7 +138,7 @@ protected String clusterName() {
      */
     protected Properties connectionProperties() {
         Properties connectionProperties = new Properties();
-        connectionProperties.put("timezone", randomKnownTimeZone());
+        connectionProperties.put(JDBC_TIMEZONE, randomKnownTimeZone());
         // in the tests, don't be lenient towards multi values
         connectionProperties.put("field.multi.value.leniency", "false");
         return connectionProperties;

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

@@ -32,6 +32,7 @@
 import java.util.Properties;
 
 import static java.util.Collections.emptyList;
+import static org.elasticsearch.xpack.sql.qa.jdbc.JdbcTestUtils.JDBC_TIMEZONE;
 
 /**
  * Tests that compare the Elasticsearch JDBC client to some other JDBC client
@@ -116,7 +117,7 @@ protected int fetchSize() {
     @Override
     protected Properties connectionProperties() {
         Properties connectionProperties = new Properties();
-        connectionProperties.setProperty("timezone", "UTC");
+        connectionProperties.setProperty(JDBC_TIMEZONE, "UTC");
         return connectionProperties;
     }
 
@@ -217,4 +218,4 @@ public interface Parser {
     public static InputStream readFromJarUrl(URL source) throws IOException {
         return source.openStream();
     }
-}
+}

x-pack/plugin/sql/qa/src/main/resources/time.csv-spec

@@ -1,26 +1,26 @@
 //
 // TIME
 //
+// All tests are run with UTC timezone
 
-// AwaitsFix: https://github.com/elastic/elasticsearch/issues/40717
-//timeExtractTimeParts
-//SELECT
-//SECOND(CAST(birth_date AS TIME)) d,
-//MINUTE(CAST(birth_date AS TIME)) m,
-//HOUR(CAST(birth_date AS TIME)) h
-//FROM "test_emp" WHERE emp_no < 10010 ORDER BY emp_no;
-//
-// d:i     | m:i      | h:i
-//0        |0         |0
-//0        |0         |0
-//0        |0         |0
-//0        |0         |0
-//0        |0         |0
-//0        |0         |0
-//0        |0         |0
-//0        |0         |0
-//0        |0         |0
-//;
+timeExtractTimeParts
+SELECT
+SECOND(CAST(birth_date AS TIME)) d,
+MINUTE(CAST(birth_date AS TIME)) m,
+HOUR(CAST(birth_date AS TIME)) h
+FROM "test_emp" WHERE emp_no < 10010 ORDER BY emp_no;
+
+ d:i     | m:i      | h:i
+0        |0         |0
+0        |0         |0
+0        |0         |0
+0        |0         |0
+0        |0         |0
+0        |0         |0
+0        |0         |0
+0        |0         |0
+0        |0         |0
+;
 
 timeAsFilter
 SELECT birth_date, last_name FROM "test_emp" WHERE birth_date::TIME = CAST('00:00:00' AS TIME) ORDER BY emp_no LIMIT 5;
@@ -59,15 +59,14 @@ null      |100445
 0         |904605
 ;
 
-// AwaitsFix: https://github.com/elastic/elasticsearch/issues/40717
-//timeAsHavingFilter
-//SELECT MINUTE_OF_HOUR(MAX(birth_date)::TIME + INTERVAL 10 MINUTES) as minute, gender FROM test_emp GROUP BY gender HAVING CAST(MAX(birth_date) AS TIME) = CAST('00:00:00.000' AS TIME) ORDER BY gender;
-//
-//minute:i    | gender:s
-//10          | null
-//10          | F
-//10          | M
-//;
+timeAsHavingFilter
+SELECT MINUTE_OF_HOUR(MAX(birth_date)::TIME + INTERVAL 10 MINUTES) as minute, gender FROM test_emp GROUP BY gender HAVING CAST(MAX(birth_date) AS TIME) = CAST('00:00:00.000' AS TIME) ORDER BY gender;
+
+minute:i    | gender:s
+10          | null
+10          | F
+10          | M
+;
 
 timeAsHavingFilterNoMatch
 SELECT MINUTE_OF_HOUR(MAX(birth_date)::TIME) as minute, gender FROM test_emp GROUP BY gender HAVING CAST(MAX(birth_date) AS TIME) > CAST('00:00:00.000' AS TIME);

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/datetime/TimeProcessor.java

@@ -16,7 +16,6 @@
 
 public class TimeProcessor extends DateTimeProcessor {
 
-
     public static final String NAME = "time";
 
     public TimeProcessor(DateTimeExtractor extractor, ZoneId zoneId) {
@@ -27,6 +26,11 @@ public TimeProcessor(StreamInput in) throws IOException {
         super(in);
     }
 
+    @Override
+    public String getWriteableName() {
+        return NAME;
+    }
+
     @Override
     public Object process(Object input) {
         if (input instanceof OffsetTime) {

x-pack/plugin/sql/src/test/java/org/elasticsearch/xpack/sql/expression/ProcessorTests.java

@@ -30,7 +30,6 @@ public static void init() throws Exception {
         processors = NodeSubclassTests.subclassesOf(Processor.class);
     }
 
-
     public void testProcessorRegistration() throws Exception {
         LinkedHashSet<String> registered = Processors.getNamedWriteables().stream()
                 .map(e -> e.name)
@@ -39,32 +38,33 @@ public void testProcessorRegistration() throws Exception {
         // discover available processors
         int missing = processors.size() - registered.size();
 
+        List<String> notRegistered = new ArrayList<>();
+        for (Class<? extends Processor> proc : processors) {
+            String procName = proc.getName();
+            assertTrue(procName + " does NOT implement NamedWriteable", NamedWriteable.class.isAssignableFrom(proc));
+            Field name = null;
+            String value = null;
+            try {
+                name = proc.getField("NAME");
+            } catch (Exception ex) {
+                fail(procName + " does NOT provide a NAME field\n" + ex);
+            }
+            try {
+                value = name.get(proc).toString();
+            } catch (Exception ex) {
+                fail(procName + " does NOT provide a static NAME field\n" + ex);
+            }
+            if (!registered.contains(value)) {
+                notRegistered.add(procName);
+            }
+            Class<?> declaringClass = proc.getMethod("getWriteableName").getDeclaringClass();
+            assertEquals("Processor: " + proc + " doesn't override getWriteableName", proc, declaringClass);
+        }
 
         if (missing > 0) {
-            List<String> notRegistered = new ArrayList<>();
-            for (Class<? extends Processor> proc : processors) {
-                String procName = proc.getName();
-                assertTrue(procName + " does NOT implement NamedWriteable", NamedWriteable.class.isAssignableFrom(proc));
-                Field name = null;
-                String value = null;
-                try {
-                    name = proc.getField("NAME");
-                } catch (Exception ex) {
-                    fail(procName + " does NOT provide a NAME field\n" + ex);
-                }
-                try {
-                    value = name.get(proc).toString();
-                } catch (Exception ex) {
-                    fail(procName + " does NOT provide a static NAME field\n" + ex);
-                }
-                if (!registered.contains(value)) {
-                    notRegistered.add(procName);
-                }
-            }
-            
             fail(missing + " processor(s) not registered : " + notRegistered);
         } else {
             assertEquals("Detection failed: discovered more registered processors than classes", 0, missing);
         }
     }
-}
+}