SQL: Add xDBC and CLI support. QA CSV specs

x-pack/plugin/ql/src/main/java/org/elasticsearch/xpack/ql/execution/search/extractor/AbstractFieldHitExtractor.java

@@ -17,10 +17,12 @@
 
 import java.io.IOException;
 import java.time.ZoneId;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 
+import static java.util.Collections.emptyList;
 import static java.util.Collections.singletonList;
 
 /**
@@ -47,7 +49,7 @@ public Object handle(Object object, String fieldName) {
         EXTRACT_ARRAY {
             @Override
             public Object handle(Object object, String _ignored) {
-                return object instanceof List ? object : singletonList(object);
+                return object == null ? emptyList() : (object instanceof List ? object : singletonList(object));
             }
         };
 
@@ -127,18 +129,10 @@ public void writeTo(StreamOutput out) throws IOException {
 
     @Override
     public Object extract(SearchHit hit) {
-        Object value = null;
-        DocumentField field = null;
-        if (hitName != null) {
-            // a nested field value is grouped under the nested parent name (ie dep.dep_name lives under "dep":[{dep_name:value}])
-            field = hit.field(hitName);
-        } else {
-            field = hit.field(fieldName);
-        }
-        if (field != null) {
-            value = unwrapFieldsMultiValue(field.getValues());
-        }
-        return value;
+        // hitName: a nested field value is grouped under the nested parent name (ie dep.dep_name lives under "dep":[{dep_name:value}])
+        String lookupName = hitName != null ? hitName : fieldName;
+        DocumentField field = hit.field(lookupName);
+        return multiValueHandling.handle(field != null ? unwrapFieldsMultiValue(field.getValues()) : null, lookupName);
     }
 
     protected Object unwrapFieldsMultiValue(Object values) {
@@ -155,11 +149,16 @@ protected Object unwrapFieldsMultiValue(Object values) {
                 return null;
             } else {
                 if (isPrimitive(list) == false) {
-                    if (list.size() == 1 || multiValueHandling == MultiValueHandling.EXTRACT_ONE) {
-                        return unwrapFieldsMultiValue(list.get(0));
-                    } else {
-                        throw new QlIllegalArgumentException("Arrays (returned by [{}]) are not supported", fieldName);
+                    List<Object> unwrappedList = new ArrayList<>();
+                    for (Object o : list) {
+                        Object unwrapped = unwrapFieldsMultiValue(o);
+                        if (unwrapped instanceof List) {
+                            unwrappedList.addAll((List<?>) unwrapped);
+                        } else {
+                            unwrappedList.add(unwrapped);
+                        }
                     }
+                    return unwrappedList;
                 }
             }
         }

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

@@ -48,7 +48,24 @@ public enum EsType implements SQLType {
     INTERVAL_MINUTE_TO_SECOND(ExtraTypes.INTERVAL_MINUTE_SECOND),
     GEO_POINT(                ExtraTypes.GEOMETRY),
     GEO_SHAPE(                ExtraTypes.GEOMETRY),
-    SHAPE(                    ExtraTypes.GEOMETRY);
+    SHAPE(                    ExtraTypes.GEOMETRY),
+    BOOLEAN_ARRAY(            Types.ARRAY),
+    BYTE_ARRAY(               Types.ARRAY),
+    SHORT_ARRAY(              Types.ARRAY),
+    INTEGER_ARRAY(            Types.ARRAY),
+    LONG_ARRAY(               Types.ARRAY),
+    DOUBLE_ARRAY(             Types.ARRAY),
+    FLOAT_ARRAY(              Types.ARRAY),
+    HALF_FLOAT_ARRAY(         Types.ARRAY),
+    SCALED_FLOAT_ARRAY(       Types.ARRAY),
+    KEYWORD_ARRAY(            Types.ARRAY),
+    TEXT_ARRAY(               Types.ARRAY),
+    DATETIME_ARRAY(           Types.ARRAY),
+    IP_ARRAY(                 Types.ARRAY),
+    BINARY_ARRAY(             Types.ARRAY),
+    GEO_SHAPE_ARRAY(          Types.ARRAY),
+    GEO_POINT_ARRAY(          Types.ARRAY),
+    SHAPE_ARRAY(              Types.ARRAY);
 
     private final Integer type;
 

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

@@ -0,0 +1,99 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.sql.jdbc;
+
+import java.sql.Array;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.SQLFeatureNotSupportedException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+import static org.elasticsearch.xpack.sql.jdbc.TypeUtils.baseType;
+
+public class JdbcArray implements Array {
+
+    private final EsType type;
+    private final List<?> values;
+
+    public JdbcArray(EsType type, List<?> values) {
+        this.type = type;
+        this.values = values;
+    }
+    @Override
+    public String getBaseTypeName() throws SQLException {
+        return baseType(type).getName();
+    }
+
+    @Override
+    public int getBaseType() throws SQLException {
+        return baseType(type).getVendorTypeNumber();
+    }
+
+    @Override
+    public Object getArray() throws SQLException {
+        return values.toArray();
+    }
+
+    @Override
+    public Object getArray(Map<String, Class<?>> map) throws SQLException {
+        if (map == null || map.isEmpty()) {
+            return getArray();
+        }
+        throw new SQLFeatureNotSupportedException("getArray with non-empty Map not supported");
+    }
+
+    @Override
+    public Object getArray(long index, int count) throws SQLException {
+        if (index < 1 || index > Integer.MAX_VALUE) {
+            throw new SQLException("Index value [" + index + "] out of range [1, " + Integer.MAX_VALUE + "]");
+        }
+        if (count < 0) {
+            throw new SQLException("Illegal negative count [" + count + "]");
+        }
+        int index0 = (int) index - 1; // 0-based index
+        int available = index0 < values.size() ? values.size() - index0 : 0;
+        return available > 0 ?
+            Arrays.copyOfRange(values.toArray(), index0, index0 + Math.min(count, available)):
+            new Object[0];
+    }
+
+    @Override
+    public Object getArray(long index, int count, Map<String, Class<?>> map) throws SQLException {
+        if (map == null || map.isEmpty()) {
+            return getArray(index, count);
+        }
+        throw new SQLFeatureNotSupportedException("getArray with non-empty Map not supported");
+    }
+
+    @Override
+    public ResultSet getResultSet() throws SQLException {
+        throw new SQLFeatureNotSupportedException("Array as ResultSet not supported");
+    }
+
+    @Override
+    public ResultSet getResultSet(Map<String, Class<?>> map) throws SQLException {
+        throw new SQLFeatureNotSupportedException("Array as ResultSet not supported");
+    }
+
+    @Override
+    public ResultSet getResultSet(long index, int count) throws SQLException {
+        throw new SQLFeatureNotSupportedException("Array as ResultSet not supported");
+    }
+
+    @Override
+    public ResultSet getResultSet(long index, int count, Map<String, Class<?>> map) throws SQLException {
+        throw new SQLFeatureNotSupportedException("Array as ResultSet not supported");
+    }
+
+    @Override
+    public void free() throws SQLException {
+        // nop
+    }
+}

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

@@ -44,6 +44,7 @@
 import static org.elasticsearch.xpack.sql.jdbc.JdbcDateUtils.timeAsMillisSinceEpoch;
 import static org.elasticsearch.xpack.sql.jdbc.JdbcDateUtils.timeAsTime;
 import static org.elasticsearch.xpack.sql.jdbc.JdbcDateUtils.timeAsTimestamp;
+import static org.elasticsearch.xpack.sql.jdbc.TypeUtils.isArray;
 
 class JdbcResultSet implements ResultSet, JdbcWrapper {
 
@@ -934,7 +935,11 @@ public Clob getClob(int columnIndex) throws SQLException {
 
     @Override
     public Array getArray(int columnIndex) throws SQLException {
-        throw new SQLFeatureNotSupportedException("Array not supported");
+        EsType type = columnType(columnIndex);
+        if (isArray(type) == false) {
+            throw new SQLException("Cannot get column [" + columnIndex + "] of type [" + type.getName() + "] as array");
+        }
+        return new JdbcArray(type, (List<?>) getObject(columnIndex));
     }
 
     @Override
@@ -954,7 +959,7 @@ public Clob getClob(String columnLabel) throws SQLException {
 
     @Override
     public Array getArray(String columnLabel) throws SQLException {
-        throw new SQLFeatureNotSupportedException("Array not supported");
+        return getArray(column(columnLabel));
     }
 
     @Override

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

@@ -27,8 +27,10 @@
 import java.time.Period;
 import java.time.ZoneOffset;
 import java.time.ZonedDateTime;
+import java.util.ArrayList;
 import java.util.Calendar;
 import java.util.GregorianCalendar;
+import java.util.List;
 import java.util.Locale;
 import java.util.function.Function;
 
@@ -46,6 +48,7 @@
 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;
+import static org.elasticsearch.xpack.sql.jdbc.TypeUtils.baseType;
 
 /**
  * Conversion utilities for conversion of JDBC types to Java type and back
@@ -263,6 +266,29 @@ static Object convert(Object v, EsType columnType, String typeString) throws SQL
                 }
             case IP:
                 return v.toString();
+            case BOOLEAN_ARRAY:
+            case BYTE_ARRAY:
+            case SHORT_ARRAY:
+            case INTEGER_ARRAY:
+            case LONG_ARRAY:
+            case DOUBLE_ARRAY:
+            case FLOAT_ARRAY:
+            case HALF_FLOAT_ARRAY:
+            case SCALED_FLOAT_ARRAY:
+            case KEYWORD_ARRAY:
+            case TEXT_ARRAY:
+            case DATETIME_ARRAY:
+            case IP_ARRAY:
+            case BINARY_ARRAY:
+            case GEO_SHAPE_ARRAY:
+            case GEO_POINT_ARRAY:
+            case SHAPE_ARRAY:
+                List<Object> values = new ArrayList<>();
+                for (Object o : (List<?>) v) {
+                    // null value expects a NULL type in the converter
+                    values.add(o == null ? null : convert(o, baseType(columnType), typeString.substring(0, "_ARRAY".length())));
+                }
+                return values;
             default:
                 throw new SQLException("Unexpected column type [" + typeString + "]");
 

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

@@ -12,6 +12,7 @@
 import java.sql.SQLFeatureNotSupportedException;
 import java.sql.SQLType;
 import java.sql.Timestamp;
+import java.sql.Types;
 import java.time.Duration;
 import java.time.LocalDateTime;
 import java.time.Period;
@@ -176,4 +177,22 @@ static EsType of(Class<? extends Object> clazz) throws SQLException {
         }
         return dataType;
     }
+
+    static EsType baseType(EsType type) throws SQLException {
+        String typeName = type.getName();
+        return isArray(type)
+            ? of(typeName.substring(0, typeName.length() - "_ARRAY".length()).toLowerCase(Locale.ROOT))
+            : type;
+    }
+
+    static EsType arrayOf(EsType type) throws SQLException {
+        if (isArray(type)) {
+            throw new SQLException("multidimentional array types not supported");
+        }
+        return ENUM_NAME_TO_TYPE.get(type.name().toLowerCase(Locale.ROOT) + "_array");
+    }
+
+    static boolean isArray(EsType type) {
+        return type.getVendorTypeNumber() == Types.ARRAY;
+    }
 }

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

@@ -0,0 +1,69 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.sql.jdbc;
+
+import org.elasticsearch.test.ESTestCase;
+
+import java.sql.Array;
+import java.sql.SQLException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static java.util.Arrays.asList;
+import static java.util.Collections.emptyList;
+import static org.elasticsearch.xpack.sql.jdbc.TypeUtils.baseType;
+
+public class JdbcArrayTests extends ESTestCase {
+
+    static final List<EsType> ARRAY_TYPES = Arrays.stream(EsType.values()).filter(TypeUtils::isArray).collect(Collectors.toList());
+
+    public void testMetaData() throws Exception {
+        for (EsType arrayType : ARRAY_TYPES) {
+            Array array = new JdbcArray(arrayType, emptyList());
+
+            assertEquals(baseType(arrayType).getVendorTypeNumber().intValue(), array.getBaseType());
+            assertEquals(baseType(arrayType).getName(), array.getBaseTypeName());
+        }
+    }
+
+    public void testGetArray() throws SQLException {
+        List<Long> expected = randomList(1, 10, ESTestCase::randomLong);
+        Array array = new JdbcArray(EsType.LONG_ARRAY, expected);
+
+        List<?> actual = asList((Object[]) array.getArray());
+        assertEquals(expected, actual);
+    }
+
+    public void testArraySlicing() throws SQLException {
+        List<Integer> values = IntStream.rangeClosed(0, 9).boxed().collect(Collectors.toList());
+        Array array = new JdbcArray(EsType.INTEGER_ARRAY, values);
+
+        Object[] empty = (Object[]) array.getArray(11, 2);
+        assertEquals(0, empty.length);
+
+        Object[] edgeSingleton = (Object[]) array.getArray(10, 2);
+        assertEquals(9, edgeSingleton[0]);
+
+        Object[] midSingleton = (Object[]) array.getArray(5, 1);
+        assertEquals(4, midSingleton[0]);
+
+        Object[] contained = (Object[]) array.getArray(4, 3);
+        assertEquals(asList(3, 4, 5), asList(contained));
+
+        Object[] overlapping = (Object[]) array.getArray(9, 3);
+        assertEquals(asList(8, 9), asList(overlapping));
+
+        SQLException sqle = expectThrows(SQLException.class, () -> array.getArray(0, 9));
+        assertEquals("Index value [0] out of range [1, 2147483647]", sqle.getMessage());
+
+        sqle = expectThrows(SQLException.class, () -> array.getArray(Integer.MAX_VALUE + 1L, 9));
+        assertEquals("Index value [2147483648] out of range [1, 2147483647]", sqle.getMessage());
+    }
+}

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

@@ -21,6 +21,7 @@
 import java.time.LocalDateTime;
 import java.time.ZonedDateTime;
 import java.util.Calendar;
+import java.util.Collections;
 import java.util.Date;
 import java.util.Locale;
 import java.util.Map;
@@ -585,6 +586,13 @@ public void testThrownExceptionsWhenSettingByteArrayValuesToEsTypes() throws SQL
         assertEquals("Conversion from type [byte[]] to [INTERVAL_DAY_TO_MINUTE] not supported", sqle.getMessage());
     }
 
+    public void testThrownExceptionWhenSettingArrayValue() throws SQLException {
+        JdbcPreparedStatement jps = createJdbcPreparedStatement();
+        JdbcArray array = new JdbcArray(LONG, Collections.emptyList());
+        SQLException sqle = expectThrows(SQLFeatureNotSupportedException.class, () -> jps.setArray(1, array));
+        assertEquals("Objects of type [java.sql.Array] are not supported", sqle.getMessage());
+    }
+
     private JdbcPreparedStatement createJdbcPreparedStatement() throws SQLException {
         return new JdbcPreparedStatement(null, JdbcConfiguration.create("jdbc:es://l:1", null, 0), "?");
     }