[SPARK-53546][TESTS][FOLLOW-UP] Fix nested array schema evolution and style for InMemoryBaseTable

### What changes were proposed in this pull request?
Fix nested array struct schema evolution case for InMemoryDataSource, also style fixes.

### Why are the changes needed?
Address style review comment (#52299 (comment)) and a small bug fix for #52299 (test-only)

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

### How was this patch tested?
Ran affected unit test

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

Closes #52359 from szehon-ho/in_memory_schema_evolution_fix.

Authored-by: Szehon Ho <szehon.apache@gmail.com>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>

Author: szehon-ho

sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala

@@ -351,8 +351,9 @@ abstract class InMemoryBaseTable(
     this
   }
 
-  def alterTableWithData(data: Array[BufferedRows],
-                         newSchema: StructType): InMemoryBaseTable = {
+  def alterTableWithData(
+      data: Array[BufferedRows],
+      newSchema: StructType): InMemoryBaseTable = {
     data.foreach { bufferedRow =>
       val oldSchema = bufferedRow.schema
       bufferedRow.rows.foreach { row =>
@@ -757,8 +758,7 @@ object InMemoryBaseTable {
  * @param key partition key
  * @param schema schema used to write the rows
  */
-class BufferedRows(val key: Seq[Any],
-                   val schema: StructType)
+class BufferedRows(val key: Seq[Any], val schema: StructType)
   extends WriterCommitMessage
     with InputPartition with HasPartitionKey with HasPartitionStatistics with Serializable {
   val log = new mutable.ArrayBuffer[InternalRow]()
@@ -892,8 +892,8 @@ private class BufferedRowsReader(
             if (arrayData == null) {
               null
             } else {
-              extractArrayValue(arrayData, elementType,
-                readSchema.fields(readIndex).dataType)
+              val writeType = writeSchema.fields(writeIndex).dataType.asInstanceOf[ArrayType]
+              extractArrayValue(arrayData, elementType, writeType.elementType)
             }
 
           case dt =>
@@ -906,19 +906,21 @@ private class BufferedRowsReader(
     }
   }
 
-  private def extractArrayValue(arrayData: ArrayData,
-                                readType: DataType,
-                                writeType: DataType): ArrayData = {
+  private def extractArrayValue(
+      arrayData: ArrayData,
+      readType: DataType,
+      writeType: DataType): ArrayData = {
     val elements = arrayData.toArray[Any](readType)
     val convertedElements = extractCollection(elements, readType, writeType)
     new GenericArrayData(convertedElements)
   }
 
-  private def extractMapValue(mapData: MapData,
-                              readKeyType: DataType,
-                              readValueType: DataType,
-                              writeKeyType: DataType,
-                              writeValueType: DataType): MapData = {
+  private def extractMapValue(
+      mapData: MapData,
+      readKeyType: DataType,
+      readValueType: DataType,
+      writeKeyType: DataType,
+      writeValueType: DataType): MapData = {
     val keys = mapData.keyArray().toArray[Any](readKeyType)
     val values = mapData.valueArray().toArray[Any](readValueType)
 
@@ -927,9 +929,10 @@ private class BufferedRowsReader(
     ArrayBasedMapData(convertedKeys, convertedValues)
   }
 
-  private def extractCollection(elements: Array[Any],
-                                readType: DataType,
-                                writeType: DataType) = {
+  private def extractCollection(
+      elements: Array[Any],
+      readType: DataType,
+      writeType: DataType) = {
     (readType, writeType) match {
       case (readSt: StructType, writeSt: StructType) =>
         elements.map { elem =>

sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoTableSuiteBase.scala

@@ -2511,7 +2511,7 @@ abstract class MergeIntoTableSuiteBase extends RowLevelOperationSuiteBase
     }
   }
 
-  test("merge into schema evolution add column with nested field and set all columns") {
+  test("merge into schema evolution add column with nested struct and set all columns") {
     Seq(true, false).foreach { withSchemaEvolution =>
       withTempView("source") {
         createAndInitTable(
@@ -2570,7 +2570,7 @@ abstract class MergeIntoTableSuiteBase extends RowLevelOperationSuiteBase
     }
   }
 
-  test("merge into schema evolution replace column with nested field and set explicit columns") {
+  test("merge into schema evolution replace column with nested struct and set explicit columns") {
     Seq(true, false).foreach { withSchemaEvolution =>
       withTempView("source") {
         createAndInitTable(
@@ -2631,7 +2631,8 @@ abstract class MergeIntoTableSuiteBase extends RowLevelOperationSuiteBase
     }
   }
 
-  // TODO- support schema evolution for missing nested types using UPDATE SET * and INSERT *
+  // currently the source struct needs to be fully compatible with target struct
+  // i.e. cannot remove a nested field
   test("merge into schema evolution replace column with nested field and set all columns") {
     Seq(true, false).foreach { withSchemaEvolution =>
       withTempView("source") {
@@ -2723,8 +2724,7 @@ abstract class MergeIntoTableSuiteBase extends RowLevelOperationSuiteBase
           sql(mergeStmt)
           checkAnswer(
             sql(s"SELECT * FROM $tableNameAsString"),
-            // TODO- InMemoryBaseTable does not return null for nested schema evolution.
-            Seq(Row(0, Array(Row(1, "a", true), Row(2, "b", true)), "sales"),
+            Seq(Row(0, Array(Row(1, "a", null), Row(2, "b", null)), "sales"),
               Row(1, Array(Row(10, "c", true), Row(20, "d", false)), "hr"),
               Row(2, Array(Row(30, "d", false), Row(40, "e", true)), "engineering")))
         } else {