[SPARK-46679][SQL] Fix for SparkUnsupportedOperationException Not found an encoder of the type T, when using Parameterized class

This pr is a rebase/continue of https://github.com/apache/spark/pull/48304/files, which was unexpectedly closed even though the issue is still there.

### What changes were proposed in this pull request?
That pr fixes a bug in JavaTypeInference.encoderFor.  In the parameterizableType case, it overrides the existing map of parameterizable type -> real type.  So any nested parameterizable type will fail.

### Why are the changes needed?
Fix nested parameterization cases for Java Encoder

### Does this PR introduce _any_ user-facing change?
No

### How was this patch tested?
The original pr adds unit test to JavaTypeInferenceSuite and JavaDatasetSuite.  The test from the original pr are simplified a little to make it obvious which one is wrapping.

### Was this patch authored or co-authored using generative AI tooling?
No

Closes #52444 from szehon-ho/parameterized_encoders.

Authored-by: Szehon Ho <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>

Author: szehon-ho

sql/api/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala

@@ -139,7 +139,9 @@ object JavaTypeInference {
       encoderFor(typeVariables(tv), seenTypeSet, typeVariables)
 
     case pt: ParameterizedType =>
-      encoderFor(pt.getRawType, seenTypeSet, JavaTypeUtils.getTypeArguments(pt).asScala.toMap)
+      val newTvs = JavaTypeUtils.getTypeArguments(pt).asScala.toMap
+      val allTvs = typeVariables ++ newTvs
+      encoderFor(pt.getRawType, seenTypeSet, allTvs)
 
     case c: Class[_] =>
       if (seenTypeSet.contains(c)) {

sql/catalyst/src/test/java/org/apache/spark/sql/catalyst/JavaTypeInferenceBeans.java

@@ -78,5 +78,61 @@ static class JavaBeanWithGenericBase extends JavaBeanWithGenerics<String, String
   static class JavaBeanWithGenericHierarchy extends JavaBeanWithGenericsABC<Integer> {
 
   }
+
+  // SPARK-46679: Test classes for nested parameterized types with multi-level inheritance
+  static class Foo<T> {
+    private T t;
+
+    public T getT() {
+      return t;
+    }
+
+    public void setT(T t) {
+      this.t = t;
+    }
+  }
+
+  static class FooWrapper<U> {
+    private Foo<U> foo;
+
+    public Foo<U> getFoo() {
+      return foo;
+    }
+
+    public void setFoo(Foo<U> foo) {
+      this.foo = foo;
+    }
+  }
+
+  static class StringFooWrapper extends FooWrapper<String> {
+  }
+
+  // SPARK-46679: Additional test classes for same type variable names at different levels
+  static class StringBarWrapper extends BarWrapper<String> {
+  }
+
+  static class BarWrapper<T> {
+    private Bar<T> bar;
+
+    public Bar<T> getBar() {
+      return bar;
+    }
+
+    public void setBar(Bar<T> bar) {
+      this.bar = bar;
+    }
+  }
+
+  static class Bar<T> {
+    private T t;
+
+    public T getT() {
+      return t;
+    }
+
+    public void setT(T t) {
+      this.t = t;
+    }
+  }
 }
 

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/JavaTypeInferenceSuite.scala

@@ -24,7 +24,7 @@ import scala.beans.{BeanProperty, BooleanBeanProperty}
 import scala.reflect.{classTag, ClassTag}
 
 import org.apache.spark.SparkFunSuite
-import org.apache.spark.sql.catalyst.JavaTypeInferenceBeans.{JavaBeanWithGenericBase, JavaBeanWithGenericHierarchy, JavaBeanWithGenericsABC}
+import org.apache.spark.sql.catalyst.JavaTypeInferenceBeans.{Bar, Foo, JavaBeanWithGenericBase, JavaBeanWithGenericHierarchy, JavaBeanWithGenericsABC, StringBarWrapper, StringFooWrapper}
 import org.apache.spark.sql.catalyst.encoders.{AgnosticEncoder, UDTCaseClass, UDTForCaseClass}
 import org.apache.spark.sql.catalyst.encoders.AgnosticEncoders._
 import org.apache.spark.sql.types.{DecimalType, MapType, Metadata, StringType, StructField, StructType}
@@ -279,4 +279,26 @@ class JavaTypeInferenceSuite extends SparkFunSuite {
       ))
     assert(encoder === expected)
   }
+
+  test("SPARK-46679: resolve generics with multi-level inheritance") {
+    val encoder = JavaTypeInference.encoderFor(classOf[StringFooWrapper])
+    val expected = JavaBeanEncoder(ClassTag(classOf[StringFooWrapper]), Seq(
+      encoderField("foo", JavaBeanEncoder(
+        ClassTag(classOf[Foo[String]]),
+        Seq(encoderField("t", StringEncoder))
+      ))
+    ))
+    assert(encoder === expected)
+  }
+
+  test("SPARK-46679: resolve generics with multi-level inheritance same type names") {
+    val encoder = JavaTypeInference.encoderFor(classOf[StringBarWrapper])
+    val expected = JavaBeanEncoder(ClassTag(classOf[StringBarWrapper]), Seq(
+      encoderField("bar", JavaBeanEncoder(
+        ClassTag(classOf[Bar[String]]),
+        Seq(encoderField("t", StringEncoder))
+      ))
+    ))
+    assert(encoder === expected)
+  }
 }

sql/core/src/test/java/test/org/apache/spark/sql/JavaDatasetSuite.java

@@ -928,6 +928,7 @@ public static class SimpleJavaBean implements Serializable {
     private List<Long> f;
     private Map<Integer, String> g;
     private Map<List<Long>, Map<String, String>> h;
+    private List<List<Long>> i;
 
     public boolean isA() {
       return a;
@@ -993,6 +994,14 @@ public void setH(Map<List<Long>, Map<String, String>> h) {
       this.h = h;
     }
 
+    public List<List<Long>> getI() {
+      return i;
+    }
+
+    public void setI(List<List<Long>> i) {
+      this.i = i;
+    }
+
     @Override
     public boolean equals(Object o) {
       if (this == o) return true;
@@ -1007,7 +1016,8 @@ public boolean equals(Object o) {
       if (!e.equals(that.e)) return false;
       if (!f.equals(that.f)) return false;
       if (!g.equals(that.g)) return false;
-      return h.equals(that.h);
+      if (!h.equals(that.h)) return false;
+      return i.equals(that.i);
 
     }
 
@@ -1021,6 +1031,7 @@ public int hashCode() {
       result = 31 * result + f.hashCode();
       result = 31 * result + g.hashCode();
       result = 31 * result + h.hashCode();
+      result = 31 * result + i.hashCode();
       return result;
     }
   }
@@ -1110,6 +1121,10 @@ public void testJavaBeanEncoder() {
     Map<List<Long>, Map<String, String>> complexMap1 = new HashMap<>();
     complexMap1.put(Arrays.asList(1L, 2L), nestedMap1);
     obj1.setH(complexMap1);
+    List<Long> nestedList1 = List.of(1L, 2L, 3L);
+    List<Long> nestedList2 = List.of(4L, 5L, 6L);
+    List<List<Long>> complexList1 = List.of(nestedList1, nestedList2);
+    obj1.setI(complexList1);
 
     SimpleJavaBean obj2 = new SimpleJavaBean();
     obj2.setA(false);
@@ -1128,6 +1143,10 @@ public void testJavaBeanEncoder() {
     Map<List<Long>, Map<String, String>> complexMap2 = new HashMap<>();
     complexMap2.put(Arrays.asList(3L, 4L), nestedMap2);
     obj2.setH(complexMap2);
+    List<Long> nestedList3 = List.of(1L, 2L, 7L);
+    List<Long> nestedList4 = List.of(4L, 5L, 8L);
+    List<List<Long>> complexList2 = List.of(nestedList3, nestedList4);
+    obj2.setI(complexList2);
 
     List<SimpleJavaBean> data = Arrays.asList(obj1, obj2);
     Dataset<SimpleJavaBean> ds = spark.createDataset(data, Encoders.bean(SimpleJavaBean.class));
@@ -1148,7 +1167,8 @@ public void testJavaBeanEncoder() {
       Arrays.asList("a", "b"),
       Arrays.asList(100L, null, 200L),
       map1,
-      complexMap1});
+      complexMap1,
+      complexList1});
     Row row2 = new GenericRow(new Object[]{
       false,
       30,
@@ -1157,7 +1177,8 @@ public void testJavaBeanEncoder() {
       Arrays.asList("x", "y"),
       Arrays.asList(300L, null, 400L),
       map2,
-      complexMap2});
+      complexMap2,
+      complexList2});
     StructType schema = new StructType()
       .add("a", BooleanType, false)
       .add("b", IntegerType, false)
@@ -1166,7 +1187,8 @@ public void testJavaBeanEncoder() {
       .add("e", createArrayType(StringType))
       .add("f", createArrayType(LongType))
       .add("g", createMapType(IntegerType, StringType))
-      .add("h",createMapType(createArrayType(LongType), createMapType(StringType, StringType)));
+      .add("h", createMapType(createArrayType(LongType), createMapType(StringType, StringType)))
+      .add("i", createArrayType(createArrayType(LongType)));
     Dataset<SimpleJavaBean> ds3 = spark.createDataFrame(Arrays.asList(row1, row2), schema)
       .as(Encoders.bean(SimpleJavaBean.class));
     Assertions.assertEquals(data, ds3.collectAsList());