[apache#2447] feat(spark-connector): support alter table AddColumn and DropColumn for spark-connector

Author: caican00

integration-test/src/test/java/com/datastrato/gravitino/integration/test/spark/SparkIT.java

@@ -284,22 +284,25 @@ void testAlterTableAddAndDeleteColumn() {
     dropTableIfExists(tableName);
 
     createSimpleTable(tableName);
-    List<SparkColumnInfo> sparkOldColumnInfos = getTableInfo(tableName).getColumns();
-    sparkOldColumnInfos.forEach(
-        sparkColumnInfo ->
-            Assertions.assertFalse("col1".equalsIgnoreCase(sparkColumnInfo.getName())));
+    String[] oldColumnNames =
+        getTableInfo(tableName).getColumns().stream()
+            .map(SparkColumnInfo::getName)
+            .toArray(String[]::new);
+    Assertions.assertEquals(oldColumnNames, new String[] {"id", "name", "age"});
 
     sql(String.format("ALTER TABLE %S ADD COLUMNS (col1 string)", tableName));
-    List<SparkColumnInfo> sparkAddColumnInfos = getTableInfo(tableName).getColumns();
-    Assertions.assertTrue(
-        sparkAddColumnInfos.stream()
-            .anyMatch(sparkColumnInfo -> "col1".equalsIgnoreCase(sparkColumnInfo.getName())));
+    String[] addColumnNames =
+        getTableInfo(tableName).getColumns().stream()
+            .map(SparkColumnInfo::getName)
+            .toArray(String[]::new);
+    Assertions.assertEquals(addColumnNames, new String[] {"id", "name", "age", "col1"});
 
     sql(String.format("ALTER TABLE %S DROP COLUMNS (col1)", tableName));
-    List<SparkColumnInfo> sparkDeleteColumnInfos = getTableInfo(tableName).getColumns();
-    sparkDeleteColumnInfos.forEach(
-        sparkColumnInfo ->
-            Assertions.assertFalse("col1".equalsIgnoreCase(sparkColumnInfo.getName())));
+    String[] deleteColumnNames =
+        getTableInfo(tableName).getColumns().stream()
+            .map(SparkColumnInfo::getName)
+            .toArray(String[]::new);
+    Assertions.assertEquals(deleteColumnNames, new String[] {"id", "name", "age"});
   }
 
   private void checkTableReadWrite(SparkTableInfo table) {

spark-connector/src/main/java/com/datastrato/gravitino/spark/connector/catalog/GravitinoCatalog.java

@@ -381,13 +381,17 @@ static com.datastrato.gravitino.rel.TableChange transformTableChange(TableChange
 
   private static com.datastrato.gravitino.rel.TableChange.ColumnPosition transformColumnPosition(
       TableChange.ColumnPosition columnPosition) {
-    if (columnPosition instanceof TableChange.First) {
+    if (null == columnPosition) {
+      return com.datastrato.gravitino.rel.TableChange.ColumnPosition.defaultPos();
+    } else if (columnPosition instanceof TableChange.First) {
       return com.datastrato.gravitino.rel.TableChange.ColumnPosition.first();
     } else if (columnPosition instanceof TableChange.After) {
       TableChange.After after = (TableChange.After) columnPosition;
       return com.datastrato.gravitino.rel.TableChange.ColumnPosition.after(after.column());
     } else {
-      return com.datastrato.gravitino.rel.TableChange.ColumnPosition.defaultPos();
+      throw new UnsupportedOperationException(
+          String.format(
+              "Unsupported table column position %s", columnPosition.getClass().getName()));
     }
   }
 }

spark-connector/src/test/java/com/datastrato/gravitino/spark/connector/catalog/TestTransformTableChange.java

@@ -5,8 +5,6 @@
 
 package com.datastrato.gravitino.spark.connector.catalog;
 
-import java.util.Arrays;
-import java.util.Locale;
 import org.apache.spark.sql.connector.catalog.TableChange;
 import org.apache.spark.sql.types.DataTypes;
 import org.junit.jupiter.api.Assertions;
@@ -41,43 +39,38 @@ void testTransformRemoveProperty() {
 
   @Test
   void testTransformAddColumn() {
-    TableChange sparkChange = TableChange.addColumn(new String[] {"col1"}, DataTypes.StringType);
+    TableChange.AddColumn sparkAddColumn =
+        (TableChange.AddColumn) TableChange.addColumn(new String[] {"col1"}, DataTypes.StringType);
     com.datastrato.gravitino.rel.TableChange gravitinoChange =
-        GravitinoCatalog.transformTableChange(sparkChange);
+        GravitinoCatalog.transformTableChange(sparkAddColumn);
 
-    TableChange.AddColumn sparkAddColumn = (TableChange.AddColumn) sparkChange;
     Assertions.assertTrue(
         gravitinoChange instanceof com.datastrato.gravitino.rel.TableChange.AddColumn);
     com.datastrato.gravitino.rel.TableChange.AddColumn gravitinoAddColumn =
         (com.datastrato.gravitino.rel.TableChange.AddColumn) gravitinoChange;
 
     Assertions.assertEquals(1, sparkAddColumn.fieldNames().length);
     Assertions.assertEquals(1, gravitinoAddColumn.fieldName().length);
-    Assertions.assertEquals(
-        Arrays.stream(sparkAddColumn.fieldNames()).findFirst(),
-        Arrays.stream(gravitinoAddColumn.fieldName()).findFirst());
-    Assertions.assertEquals(
-        sparkAddColumn.dataType().typeName().toLowerCase(Locale.ROOT),
-        gravitinoAddColumn.getDataType().simpleString().toLowerCase(Locale.ROOT));
+    Assertions.assertEquals(sparkAddColumn.fieldNames(), gravitinoAddColumn.fieldName());
+    Assertions.assertTrue(
+        "string".equalsIgnoreCase(gravitinoAddColumn.getDataType().simpleString()));
   }
 
   @Test
   void testTransformDeleteColumn() {
-    TableChange sparkChange = TableChange.deleteColumn(new String[] {"col1"}, true);
+    TableChange.DeleteColumn sparkDeleteColumn =
+        (TableChange.DeleteColumn) TableChange.deleteColumn(new String[] {"col1"}, true);
     com.datastrato.gravitino.rel.TableChange gravitinoChange =
-        GravitinoCatalog.transformTableChange(sparkChange);
+        GravitinoCatalog.transformTableChange(sparkDeleteColumn);
 
-    TableChange.DeleteColumn sparkDeleteColumn = (TableChange.DeleteColumn) sparkChange;
     Assertions.assertTrue(
         gravitinoChange instanceof com.datastrato.gravitino.rel.TableChange.DeleteColumn);
     com.datastrato.gravitino.rel.TableChange.DeleteColumn gravitinoDeleteColumn =
         (com.datastrato.gravitino.rel.TableChange.DeleteColumn) gravitinoChange;
 
     Assertions.assertEquals(1, sparkDeleteColumn.fieldNames().length);
     Assertions.assertEquals(1, gravitinoDeleteColumn.fieldName().length);
-    Assertions.assertEquals(
-        Arrays.stream(sparkDeleteColumn.fieldNames()).findFirst(),
-        Arrays.stream(gravitinoDeleteColumn.fieldName()).findFirst());
+    Assertions.assertEquals(sparkDeleteColumn.fieldNames(), gravitinoDeleteColumn.fieldName());
     Assertions.assertEquals(sparkDeleteColumn.ifExists(), gravitinoDeleteColumn.getIfExists());
   }
 }