[SPARK-55696][SQL] Add explicit error to Encoders.bean for interface class

common/utils/src/main/resources/error/error-conditions.json

@@ -9484,6 +9484,11 @@
       "Primitive types are not supported."
     ]
   },
+  "_LEGACY_ERROR_TEMP_2231" : {
+    "message" : [
+      "Bean encoder does not support interface types: <className>."
+    ]
+  },
   "_LEGACY_ERROR_TEMP_2233" : {
     "message" : [
       "Only Data Sources providing FileFormat are supported: <providingClass>."

sql/api/src/main/scala/org/apache/spark/sql/Encoders.scala

@@ -195,7 +195,7 @@ object Encoders {
   /**
    * Creates an encoder for Java Bean of type T.
    *
-   * T must be publicly accessible.
+   * T must be a concrete class (not an interface), and must be publicly accessible.
    *
    * supported types for java bean field:
    *   - primitive types: boolean, int, double, etc.

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

@@ -26,6 +26,7 @@ import scala.reflect.ClassTag
 
 import org.apache.commons.lang3.reflect.{TypeUtils => JavaTypeUtils}
 
+import org.apache.spark.SparkUnsupportedOperationException
 import org.apache.spark.sql.catalyst.encoders.AgnosticEncoder
 import org.apache.spark.sql.catalyst.encoders.AgnosticEncoders.{ArrayEncoder, BinaryEncoder, BoxedBooleanEncoder, BoxedByteEncoder, BoxedDoubleEncoder, BoxedFloatEncoder, BoxedIntEncoder, BoxedLongEncoder, BoxedShortEncoder, DayTimeIntervalEncoder, DEFAULT_GEOGRAPHY_ENCODER, DEFAULT_GEOMETRY_ENCODER, DEFAULT_JAVA_DECIMAL_ENCODER, EncoderField, IterableEncoder, JavaBeanEncoder, JavaBigIntEncoder, JavaEnumEncoder, LocalDateTimeEncoder, LocalTimeEncoder, MapEncoder, PrimitiveBooleanEncoder, PrimitiveByteEncoder, PrimitiveDoubleEncoder, PrimitiveFloatEncoder, PrimitiveIntEncoder, PrimitiveLongEncoder, PrimitiveShortEncoder, STRICT_DATE_ENCODER, STRICT_INSTANT_ENCODER, STRICT_LOCAL_DATE_ENCODER, STRICT_TIMESTAMP_ENCODER, StringEncoder, UDTEncoder, YearMonthIntervalEncoder}
 import org.apache.spark.sql.errors.ExecutionErrors
@@ -151,6 +152,13 @@ object JavaTypeInference {
       if (seenTypeSet.contains(c)) {
         throw ExecutionErrors.cannotHaveCircularReferencesInBeanClassError(c)
       }
+      // Encoders for interfaces are not supported because de-serialization uses its
+      // Deserializer to instantiate the class, which will not work for interfaces.
+      if (c.isInterface) {
+        throw new SparkUnsupportedOperationException(
+          errorClass = "_LEGACY_ERROR_TEMP_2231",
+          messageParameters = Map("className" -> c.getName))
+      }
 
       // TODO: we should only collect properties that have getter and setter. However, some tests
       //   pass in scala case class as java bean class which doesn't have getter and setter.

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

@@ -1918,6 +1918,117 @@ public void testReadOnlyPropertyBean() {
 
   }
 
+  /**
+   * Interface with JavaBean-style getters/setters for testing encoder with interface type.
+   */
+  public interface BeanInterface extends Serializable {
+    String getValue();
+    void setValue(String value);
+    int getId();
+    void setId(int id);
+  }
+
+  public static class BeanImplA implements BeanInterface {
+    private String value;
+    private int id;
+
+    @Override
+    public String getValue() { return value; }
+    @Override
+    public void setValue(String value) { this.value = value; }
+    @Override
+    public int getId() { return id; }
+    @Override
+    public void setId(int id) { this.id = id; }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) return true;
+      if (!(o instanceof BeanImplA)) return false;
+      BeanImplA that = (BeanImplA) o;
+      return id == that.id && Objects.equals(value, that.value);
+    }
+    @Override
+    public int hashCode() { return Objects.hash(value, id); }
+  }
+
+  public static class BeanImplB implements BeanInterface {
+    private String value;
+    private int id;
+
+    @Override
+    public String getValue() { return value; }
+    @Override
+    public void setValue(String value) { this.value = value; }
+    @Override
+    public int getId() { return id; }
+    @Override
+    public void setId(int id) { this.id = id; }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) return true;
+      if (!(o instanceof BeanImplB)) return false;
+      BeanImplB that = (BeanImplB) o;
+      return id == that.id && Objects.equals(value, that.value);
+    }
+    @Override
+    public int hashCode() { return Objects.hash(value, id); }
+  }
+
+  @Test
+  public void testBeanEncoderRejectsInterface() {
+    // Bean encoder does not support interface types (deserialization would fail).
+    Exception e = Assertions.assertThrows(Exception.class,
+        () -> Encoders.bean(BeanInterface.class));
+    Assertions.assertTrue(e.getMessage() != null && e.getMessage().contains("interface"),
+        "Expected message about interface not supported: " + e.getMessage());
+  }
+
+  @Test
+  public void testKryoEncoderWithInterface() {
+    BeanImplA a = new BeanImplA();
+    a.setValue("a");
+    a.setId(1);
+    BeanImplB b = new BeanImplB();
+    b.setValue("b");
+    b.setId(2);
+    List<BeanInterface> data = Arrays.asList(a, b);
+
+    Encoder<BeanInterface> kryoEncoder = Encoders.kryo(BeanInterface.class);
+    Dataset<BeanInterface> ds = spark.createDataset(data, kryoEncoder);
+    List<BeanInterface> collected = ds.collectAsList();
+    Assertions.assertEquals(2, collected.size());
+    Assertions.assertEquals("a", collected.get(0).getValue());
+    Assertions.assertEquals(1, collected.get(0).getId());
+    Assertions.assertEquals("b", collected.get(1).getValue());
+    Assertions.assertEquals(2, collected.get(1).getId());
+    Assertions.assertInstanceOf(BeanImplA.class, collected.get(0));
+    Assertions.assertInstanceOf(BeanImplB.class, collected.get(1));
+  }
+
+  @Test
+  public void testJavaSerializationEncoderWithInterface() {
+    BeanImplA a = new BeanImplA();
+    a.setValue("a");
+    a.setId(1);
+    BeanImplB b = new BeanImplB();
+    b.setValue("b");
+    b.setId(2);
+    List<BeanInterface> data = Arrays.asList(a, b);
+
+    Encoder<BeanInterface> javaEncoder = Encoders.javaSerialization(BeanInterface.class);
+    Dataset<BeanInterface> ds = spark.createDataset(data, javaEncoder);
+    List<BeanInterface> collected = ds.collectAsList();
+    Assertions.assertEquals(2, collected.size());
+    Assertions.assertEquals("a", collected.get(0).getValue());
+    Assertions.assertEquals(1, collected.get(0).getId());
+    Assertions.assertEquals("b", collected.get(1).getValue());
+    Assertions.assertEquals(2, collected.get(1).getId());
+    Assertions.assertInstanceOf(BeanImplA.class, collected.get(0));
+    Assertions.assertInstanceOf(BeanImplB.class, collected.get(1));
+  }
+
   public class CircularReference1Bean implements Serializable {
     private CircularReference2Bean child;