Support read multi-values from OpenSearch if no codegen triggered (#5015) (#5023)

* Support read multi-values from OpenSearch



* Fix tests



* remove reflection way



---------


(cherry picked from commit 98516d6)

Signed-off-by: Lantao Jin <ltjin@amazon.com>

Author: LantaoJin

core/src/main/java/org/opensearch/sql/data/model/ExprValueUtils.java

@@ -120,7 +120,9 @@ public static ExprValue fromObjectValue(Object o) {
     if (null == o) {
       return LITERAL_NULL;
     }
-    if (o instanceof Map) {
+    if (o instanceof ExprValue) {
+      return (ExprValue) o;
+    } else if (o instanceof Map) {
       return tupleValue((Map) o);
     } else if (o instanceof List) {
       return collectionValue(((List) o));

core/src/test/java/org/opensearch/sql/data/model/ExprValueUtilsTest.java

@@ -221,10 +221,8 @@ public void unSupportedObject() {
     Exception exception =
         assertThrows(
             ExpressionEvaluationException.class,
-            () -> ExprValueUtils.fromObjectValue(integerValue(1)));
-    assertEquals(
-        "unsupported object " + "class org.opensearch.sql.data.model.ExprIntegerValue",
-        exception.getMessage());
+            () -> ExprValueUtils.fromObjectValue(new Object()));
+    assertEquals("unsupported object class java.lang.Object", exception.getMessage());
   }
 
   @Test

integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLBasicIT.java

@@ -599,8 +599,8 @@ public void testNumericLiteral() throws IOException {
         schema("floatLiteral", "float"));
     verifyDataRows(
         result,
-        rows("hello", 20, 0.05, 0.049999999999999996, 0.05),
-        rows("world", 30, 0.05, 0.049999999999999996, 0.05));
+        rows("hello", 20, 0.05, 0.049999999999999996, 0.049999999999999996),
+        rows("world", 30, 0.05, 0.049999999999999996, 0.049999999999999996));
   }
 
   @Test

integ-test/src/yamlRestTest/resources/rest-api-spec/test/issues/4173.yml

@@ -0,0 +1,69 @@
+setup:
+  - do:
+      indices.create:
+        index: array-test
+        body:
+          settings:
+            number_of_shards: 1
+          mappings:
+            properties:
+              array_keyword:
+                type: keyword
+              array_text:
+                type: text
+              array_num:
+                type: long
+              array_boolean:
+                type: boolean
+              array_date:
+                type: date
+              num:
+                type: integer
+              parent:
+                type: object
+                properties:
+                  child_array_num:
+                    type: float
+                  child_array_keyword:
+                    type: keyword
+                  child_num:
+                    type: integer
+  - do:
+      bulk:
+        index: array-test
+        refresh: true
+        body:
+          - '{"index": {}}'
+          - '{"array_text": ["Jane Smith","hello world"],"array_keyword": [ "c++", "java" ],"array_num": [1, 2],"array_boolean": [true, false],"array_date": ["2023-01-02T22:03:34.000Z","2023-01-02T22:03:35.000Z"],"num" : 10,"parent": {"child_array_num": [1.1, 2.2], "child_array_keyword":["a a a", "b b b"], "child_num": 100}}'
+          - '{"index": {}}'
+          - '{"array_text": "OpenSearch PPL","array_keyword": "python","array_num": 3,"array_boolean": true,"array_date": "2023-01-02T22:03:36.000Z","num" : 11,"parent": {"child_array_num": 3.3, "child_array_keyword":"c c c", "child_num": 101}}'
+  - do:
+      query.settings:
+        body:
+          transient:
+            plugins.calcite.enabled : true
+---
+teardown:
+  - do:
+      query.settings:
+        body:
+          transient:
+            plugins.calcite.enabled : false
+
+---
+"Handle multiple values (array) documents":
+  - skip:
+      features:
+        - headers
+        - allowed_warnings
+  - do:
+      allowed_warnings:
+        - 'Loading the fielddata on the _id field is deprecated and will be removed in future versions. If you require sorting or aggregating on this field you should also include the id in the body of your documents, and map this field as a keyword field that has [doc_values] enabled'
+      headers:
+        Content-Type: 'application/json'
+      ppl:
+        body:
+          query: 'source=array-test '
+  - match: {"total": 2}
+  - match: {"schema": [{"name": "parent", "type": "struct"},{"name":"array_boolean","type":"boolean"},{"name":"array_num","type":"bigint"},{"name":"array_date","type":"timestamp"},{"name":"num","type":"int"},{"name":"array_text","type":"string"},{"name":"array_keyword","type":"string"}]}
+  - match: {"datarows": [[{"child_array_keyword":["a a a","b b b"],"child_num":100,"child_array_num":[1.1,2.2]},[true, false],[1,2],["2023-01-02 22:03:34","2023-01-02 22:03:35"],10,["Jane Smith","hello world"],["c++","java"]],[{"child_num": 101,"child_array_keyword":"c c c","child_array_num":3.3},true,3,"2023-01-02 22:03:36",11,"OpenSearch PPL","python"]]}

opensearch/src/main/java/org/opensearch/sql/opensearch/executor/OpenSearchExecutionEngine.java

@@ -13,6 +13,7 @@
 import java.sql.ResultSetMetaData;
 import java.sql.SQLException;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -38,6 +39,7 @@
 import org.apache.calcite.sql.validate.SqlUserDefinedFunction;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.locationtech.jts.geom.Point;
 import org.opensearch.sql.ast.statement.Explain.ExplainFormat;
 import org.opensearch.sql.calcite.CalcitePlanContext;
 import org.opensearch.sql.calcite.utils.CalciteToolsHelper.OpenSearchRelRunners;
@@ -46,6 +48,7 @@
 import org.opensearch.sql.common.response.ResponseListener;
 import org.opensearch.sql.data.model.ExprTupleValue;
 import org.opensearch.sql.data.model.ExprValue;
+import org.opensearch.sql.data.model.ExprValueUtils;
 import org.opensearch.sql.data.type.ExprCoreType;
 import org.opensearch.sql.data.type.ExprType;
 import org.opensearch.sql.executor.ExecutionContext;
@@ -55,14 +58,14 @@
 import org.opensearch.sql.executor.pagination.PlanSerializer;
 import org.opensearch.sql.expression.function.PPLFuncImpTable;
 import org.opensearch.sql.opensearch.client.OpenSearchClient;
+import org.opensearch.sql.opensearch.data.value.OpenSearchExprGeoPointValue;
 import org.opensearch.sql.opensearch.executor.protector.ExecutionProtector;
 import org.opensearch.sql.opensearch.functions.DistinctCountApproxAggFunction;
 import org.opensearch.sql.expression.function.BuiltinFunctionName;
 import org.opensearch.sql.expression.function.PPLFuncImpTable;
 import org.opensearch.sql.opensearch.client.OpenSearchClient;
 import org.opensearch.sql.opensearch.executor.protector.ExecutionProtector;
 import org.opensearch.sql.opensearch.functions.GeoIpFunction;
-import org.opensearch.sql.opensearch.util.JdbcOpenSearchDataTypeConvertor;
 import org.opensearch.sql.planner.physical.PhysicalPlan;
 import org.opensearch.sql.storage.TableScanOperator;
 import org.opensearch.client.node.NodeClient;
@@ -231,6 +234,38 @@ public void execute(
                     }));
   }
 
+  /**
+   * Process values recursively, handling geo points and nested maps. Geo points are converted to
+   * OpenSearchExprGeoPointValue. Maps are recursively processed to handle nested structures.
+   */
+  private static Object processValue(Object value) {
+    if (value == null) {
+      return null;
+    }
+    if (value instanceof Point) {
+      Point point = (Point) value;
+      return new OpenSearchExprGeoPointValue(point.getY(), point.getX());
+    }
+    if (value instanceof Map) {
+      Map<String, Object> map = (Map<String, Object>) value;
+      Map<String, Object> convertedMap = new HashMap<>();
+      for (Map.Entry<String, Object> entry : map.entrySet()) {
+        convertedMap.put(entry.getKey(), processValue(entry.getValue()));
+      }
+      return convertedMap;
+    }
+    if (value instanceof List) {
+      List<Object> list = (List<Object>) value;
+      List<Object> convertedList = new ArrayList<>();
+      for (Object item : list) {
+        convertedList.add(processValue(item));
+      }
+      return convertedList;
+    }
+    // For other types, return as-is
+    return value;
+  }
+
   private void buildResultSet(
       ResultSet resultSet,
       RelDataType rowTypes,
@@ -249,11 +284,9 @@ private void buildResultSet(
       // Loop through each column
       for (int i = 1; i <= columnCount; i++) {
         String columnName = metaData.getColumnName(i);
-        int sqlType = metaData.getColumnType(i);
-        RelDataType fieldType = fieldTypes.get(i - 1);
-        ExprValue exprValue =
-            JdbcOpenSearchDataTypeConvertor.getExprValueFromSqlType(
-                resultSet, i, sqlType, fieldType, columnName);
+        Object value = resultSet.getObject(columnName);
+        Object converted = processValue(value);
+        ExprValue exprValue = ExprValueUtils.fromObjectValue(converted);
         row.put(columnName, exprValue);
       }
       values.add(ExprTupleValue.fromExprValueMap(row));

opensearch/src/main/java/org/opensearch/sql/opensearch/util/JdbcOpenSearchDataTypeConvertor.java

@@ -1,177 +0,0 @@
-/*
- * Copyright OpenSearch Contributors
- * SPDX-License-Identifier: Apache-2.0
- */
-
-package org.opensearch.sql.opensearch.util;
-
-import java.sql.Array;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.sql.Types;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.Map;
-import lombok.experimental.UtilityClass;
-import org.locationtech.jts.geom.Point;
-import org.apache.calcite.avatica.util.ArrayImpl;
-import org.apache.calcite.rel.type.RelDataType;
-import org.apache.calcite.sql.type.SqlTypeName;
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
-import org.locationtech.jts.geom.Point;
-import org.opensearch.sql.calcite.type.ExprJavaType;
-import org.opensearch.sql.data.model.ExprDateValue;
-import org.opensearch.sql.data.model.ExprNullValue;
-import org.opensearch.sql.data.model.ExprTimeValue;
-import org.opensearch.sql.data.model.ExprTimestampValue;
-import org.opensearch.sql.data.model.ExprValue;
-import org.opensearch.sql.data.model.ExprValueUtils;
-import org.opensearch.sql.data.type.ExprCoreType;
-import org.opensearch.sql.data.type.ExprType;
-import org.opensearch.sql.opensearch.data.value.OpenSearchExprGeoPointValue;
-
-/** This class is used to convert the data type from JDBC to OpenSearch data type. */
-@UtilityClass
-public class JdbcOpenSearchDataTypeConvertor {
-  private static final Logger LOG = LogManager.getLogger();
-
-  public static ExprType getExprTypeFromSqlType(int sqlType) {
-    switch (sqlType) {
-      case Types.INTEGER:
-        return ExprCoreType.INTEGER;
-      case Types.BIGINT:
-        return ExprCoreType.LONG;
-      case Types.DOUBLE:
-      case Types.DECIMAL:
-      case Types.NUMERIC:
-        return ExprCoreType.DOUBLE;
-      case Types.FLOAT:
-        return ExprCoreType.FLOAT;
-      case Types.DATE:
-        return ExprCoreType.DATE;
-      case Types.TIMESTAMP:
-        return ExprCoreType.TIMESTAMP;
-      case Types.BOOLEAN:
-        return ExprCoreType.BOOLEAN;
-      case Types.VARCHAR:
-      case Types.CHAR:
-      case Types.LONGVARCHAR:
-        return ExprCoreType.STRING;
-      default:
-        // TODO unchecked OpenSearchDataType
-        return ExprCoreType.UNKNOWN;
-    }
-  }
-
-  public static ExprValue getExprValueFromSqlType(
-      ResultSet rs, int i, int sqlType, RelDataType fieldType, String fieldName)
-      throws SQLException {
-    Object value = rs.getObject(i);
-    if (value == null) {
-      return ExprNullValue.of();
-    }
-
-    if (fieldType instanceof ExprJavaType && value instanceof ExprValue) {
-      return (ExprValue) value;
-    } else if (fieldType.getSqlTypeName() == SqlTypeName.GEOMETRY) {
-      // Use getObject by name instead of index to avoid Avatica's transformation on the accessor.
-      // Otherwise, Avatica will transform Geometry to String.
-      Point geoPoint = (Point) rs.getObject(fieldName);
-      return new OpenSearchExprGeoPointValue(geoPoint.getY(), geoPoint.getX());
-    }
-
-    try {
-      switch (sqlType) {
-        case Types.VARCHAR:
-        case Types.CHAR:
-        case Types.LONGVARCHAR:
-          return ExprValueUtils.fromObjectValue(rs.getString(i));
-
-        case Types.INTEGER:
-          return ExprValueUtils.fromObjectValue(rs.getInt(i));
-
-        case Types.BIGINT:
-          return ExprValueUtils.fromObjectValue(rs.getLong(i));
-
-        case Types.FLOAT:
-        case Types.REAL:
-          return ExprValueUtils.fromObjectValue(rs.getFloat(i));
-
-        case Types.DECIMAL:
-        case Types.NUMERIC:
-        case Types.DOUBLE:
-          return ExprValueUtils.fromObjectValue(rs.getDouble(i));
-
-        case Types.DATE:
-          String dateStr = rs.getString(i);
-          return new ExprDateValue(dateStr);
-
-        case Types.TIME:
-          String timeStr = rs.getString(i);
-          return new ExprTimeValue(timeStr);
-
-        case Types.TIMESTAMP:
-          String timestampStr = rs.getString(i);
-          return new ExprTimestampValue(timestampStr);
-
-        case Types.BOOLEAN:
-          return ExprValueUtils.fromObjectValue(rs.getBoolean(i));
-
-        case Types.ARRAY:
-          Array array = rs.getArray(i);
-          if (array instanceof ArrayImpl) {
-            return ExprValueUtils.fromObjectValue(
-                Arrays.asList((Object[]) ((ArrayImpl) value).getArray()));
-          }
-          return ExprValueUtils.fromObjectValue(array);
-
-        default:
-          LOG.debug(
-              "Unchecked sql type: {}, return Object type {}",
-              sqlType,
-              value.getClass().getTypeName());
-          return convertComplexValue(value);
-      }
-    } catch (SQLException e) {
-      LOG.error("Error converting SQL type {}: {}", sqlType, e.getMessage());
-      throw e;
-    }
-  }
-
-  /**
-   * Convert complex values like Maps that may contain geo points. This method recursively processes
-   * Maps to handle nested geo points and converts them to appropriate ExprValue representations.
-   */
-  private static ExprValue convertComplexValue(Object value) {
-    Object converted = processValue(value);
-    return ExprValueUtils.fromObjectValue(converted);
-  }
-
-  /**
-   * Process values recursively, handling geo points and nested maps. Geo points are converted to
-   * OpenSearchExprGeoPointValue. Maps are recursively processed to handle nested structures.
-   */
-  private static Object processValue(Object value) {
-    if (value == null) {
-      return null;
-    }
-
-    if (value instanceof Point) {
-      Point point = (Point) value;
-      return new OpenSearchExprGeoPointValue(point.getY(), point.getX());
-    }
-
-    if (value instanceof Map) {
-      Map<String, Object> map = (Map<String, Object>) value;
-      Map<String, Object> convertedMap = new HashMap<>();
-      for (Map.Entry<String, Object> entry : map.entrySet()) {
-        convertedMap.put(entry.getKey(), processValue(entry.getValue()));
-      }
-      return convertedMap;
-    }
-
-    // For other types, return as-is
-    return value;
-  }
-}