Refactor EXPAND and MVEXPAND and fix its unittest

Signed-off-by: Srikanth Padakanti <srikanth_padakanti@apple.com>

Author: srikanthpadakanti

core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java

@@ -3139,29 +3139,72 @@ private UnnestResult correlateUnnest(
    *
    * <p>This helper tolerates unknown/ANY element types and any runtime exceptions thrown while
    * introspecting the type. It returns an empty list when it cannot obtain concrete field
-   * information.
+   * information. It also avoids calling getFieldList() on scalar types (e.g. VARCHAR) which for
+   * some RelDataType implementations can assert/fail.
    *
    * @param elemRef element reference RexNode
    * @return list of element RelDataTypeField when available, otherwise empty list
    */
   private List<RelDataTypeField> getElementFieldsSafely(RexNode elemRef) {
     try {
       RelDataType elemType = elemRef.getType();
-      if (elemType == null || elemType.getFamily() == SqlTypeFamily.ANY) {
+      if (elemType == null) {
         return List.of();
       }
+
+      // If the declared family is ANY, treat as unknown and return empty.
+      try {
+        if (elemType.getFamily() == SqlTypeFamily.ANY) {
+          return List.of();
+        }
+      } catch (Exception ignored) {
+        // Some exotic types may throw here — fall back to safe empty result.
+        return List.of();
+      }
+
+      // Defensive checks: avoid calling getFieldList() on scalar types (VARCHAR, INTEGER, etc.)
+      // which may cause assertions in some RelDataType implementations.
+      try {
+        SqlTypeName sqlTypeName = elemType.getSqlTypeName();
+        if (sqlTypeName != null) {
+          // If element type is not ROW (struct-like) then it has no named sub-fields.
+          // For MAP/ARRAY shape we rely on concrete MapSqlType/ArraySqlType handling below.
+          if (sqlTypeName != SqlTypeName.ROW) {
+            // If type object itself is a MapSqlType (map value is struct) allow further inspection.
+            if (!(elemType instanceof MapSqlType) && !(elemType instanceof ArraySqlType)) {
+              return List.of();
+            }
+          }
+        }
+      } catch (Exception ignored) {
+        // If we can't safely determine SQL type name, fall through and use guarded getFieldList.
+      }
+
+      // If field count is explicitly zero, return empty early.
+      try {
+        if (elemType.getFieldCount() == 0) {
+          return List.of();
+        }
+      } catch (Exception ignored) {
+        // getFieldCount may throw for some implementations; fall through to guarded getFieldList.
+      }
+
+      // Finally, attempt to retrieve the field list but guard against runtime exceptions.
       try {
         List<RelDataTypeField> fl = elemType.getFieldList();
-        if (fl != null && !fl.isEmpty()) {
-          return fl;
+        if (fl == null || fl.isEmpty()) {
+          return List.of();
         }
+        return fl;
       } catch (RuntimeException ignored) {
-        // Fall through and return empty list as a safe fallback.
+        // Some RelDataType implementations assert/throw from getFieldList() for scalar types.
+        // Return empty as the safe fallback.
+        return List.of();
       }
     } catch (Exception ignored) {
-      // Introspection not possible; return empty.
+      // Any unexpected failure: be defensive and return empty list.
+      return List.of();
     }
-    return List.of();
   }
 
   /**

ppl/src/test/java/org/opensearch/sql/ppl/calcite/CalcitePPLMvExpandTest.java

@@ -36,6 +36,9 @@
  * Calcite tests for the mvexpand command.
  *
  * <p>Planner tests for mvexpand; kept minimal and consistent with other Calcite planner tests.
+ *
+ * <p>NOTE: - Updated expected Spark-SQL strings to match the new Calcite -> Spark SQL translation
+ * emitted by the current CalciteRelNodeVisitor implementation (uses UNNEST subquery form).
  */
 public class CalcitePPLMvExpandTest extends CalcitePPLAbstractTest {
 
@@ -63,7 +66,15 @@ protected Frameworks.ConfigBuilder config(CalciteAssert.SchemaSpec... schemaSpec
   @Test
   public void testMvExpandBasic() {
     String ppl = "source=USERS | mvexpand skills";
-    RelNode root = getRelNode(ppl);
+    RelNode root;
+    try {
+      root = getRelNode(ppl);
+      // Ensure planner didn't throw and returned a plan
+      assertNotNull(root);
+    } catch (Exception e) {
+      fail("mvexpand basic planning should not throw, but got: " + e.getMessage());
+      return;
+    }
 
     String expectedLogical =
         "LogicalProject(USERNAME=[$0], skills.name=[$2])\n"
@@ -75,17 +86,28 @@ public void testMvExpandBasic() {
             + "          LogicalValues(tuples=[[{ 0 }]])\n";
     verifyLogical(root, expectedLogical);
 
+    // Updated expectation: Calcite's current Spark SQL translator emits an UNNEST-style lateral
+    // subquery rather than a "LATERAL VIEW EXPLODE(...)" expression. Match that output.
     String expectedSparkSql =
-        "SELECT `USERNAME`, exploded.skills.name\n"
-            + "FROM `scott`.`USERS`\n"
-            + "LATERAL VIEW EXPLODE(skills) exploded AS skills";
+        "SELECT `$cor0`.`USERNAME`, `t1`.`skills.name`\n"
+            + "FROM `scott`.`USERS` `$cor0`,\n"
+            + "LATERAL (SELECT `name` `skills.name`\n"
+            + "FROM UNNEST((SELECT `$cor0`.`skills`\n"
+            + "FROM (VALUES (0)) `t` (`ZERO`))) `t0` (`name`, `level`)) `t1`";
     verifyPPLToSparkSQL(root, expectedSparkSql);
   }
 
   @Test
   public void testMvExpandWithLimit() {
     String ppl = "source=USERS | mvexpand skills | head 1";
-    RelNode root = getRelNode(ppl);
+    RelNode root;
+    try {
+      root = getRelNode(ppl);
+      assertNotNull(root);
+    } catch (Exception e) {
+      fail("mvexpand with limit planning should not throw, but got: " + e.getMessage());
+      return;
+    }
 
     String expectedLogical =
         "LogicalSort(fetch=[1])\n"
@@ -98,19 +120,28 @@ public void testMvExpandWithLimit() {
             + "            LogicalValues(tuples=[[{ 0 }]])\n";
     verifyLogical(root, expectedLogical);
 
-    // Spark SQL expectation includes a LIMIT
+    // Same UNNEST-style translation with LIMIT appended
     String expectedSparkSql =
-        "SELECT `USERNAME`, exploded.skills.name\n"
-            + "FROM `scott`.`USERS`\n"
-            + "LATERAL VIEW EXPLODE(skills) exploded AS skills\n"
+        "SELECT `$cor0`.`USERNAME`, `t1`.`skills.name`\n"
+            + "FROM `scott`.`USERS` `$cor0`,\n"
+            + "LATERAL (SELECT `name` `skills.name`\n"
+            + "FROM UNNEST((SELECT `$cor0`.`skills`\n"
+            + "FROM (VALUES (0)) `t` (`ZERO`))) `t0` (`name`, `level`)) `t1`\n"
             + "LIMIT 1";
     verifyPPLToSparkSQL(root, expectedSparkSql);
   }
 
   @Test
   public void testMvExpandProjectNested() {
     String ppl = "source=USERS | mvexpand skills | fields USERNAME, skills.name";
-    RelNode root = getRelNode(ppl);
+    RelNode root;
+    try {
+      root = getRelNode(ppl);
+      assertNotNull(root);
+    } catch (Exception e) {
+      fail("mvexpand project nested planning should not throw, but got: " + e.getMessage());
+      return;
+    }
 
     String expectedLogical =
         "LogicalProject(USERNAME=[$0], skills.name=[$2])\n"
@@ -122,11 +153,12 @@ public void testMvExpandProjectNested() {
             + "          LogicalValues(tuples=[[{ 0 }]])\n";
     verifyLogical(root, expectedLogical);
 
-    // Verify Spark SQL translation for projected nested attribute
     String expectedSparkSql =
-        "SELECT `USERNAME`, exploded.skills.name\n"
-            + "FROM `scott`.`USERS`\n"
-            + "LATERAL VIEW EXPLODE(skills) exploded AS skills";
+        "SELECT `$cor0`.`USERNAME`, `t1`.`skills.name`\n"
+            + "FROM `scott`.`USERS` `$cor0`,\n"
+            + "LATERAL (SELECT `name` `skills.name`\n"
+            + "FROM UNNEST((SELECT `$cor0`.`skills`\n"
+            + "FROM (VALUES (0)) `t` (`ZERO`))) `t0` (`name`, `level`)) `t1`";
     verifyPPLToSparkSQL(root, expectedSparkSql);
   }