[#4412] feat(api): add validate check for default position when updating column position

api/src/main/java/org/apache/gravitino/rel/TableChange.java

@@ -20,6 +20,7 @@
 
 package org.apache.gravitino.rel;
 
+import com.google.common.base.Preconditions;
 import java.util.Arrays;
 import java.util.Objects;
 import org.apache.gravitino.annotation.Evolving;
@@ -1353,6 +1354,8 @@ final class UpdateColumnPosition implements ColumnChange {
     private final ColumnPosition position;
 
     private UpdateColumnPosition(String[] fieldName, ColumnPosition position) {
+      Preconditions.checkArgument(
+          position != ColumnPosition.defaultPos(), "Position cannot be DEFAULT");
       this.fieldName = fieldName;
       this.position = position;
     }

api/src/test/java/org/apache/gravitino/TestTableChange.java

@@ -43,6 +43,7 @@
 import org.apache.gravitino.rel.expressions.literals.Literals;
 import org.apache.gravitino.rel.types.Type;
 import org.apache.gravitino.rel.types.Types;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 public class TestTableChange {
@@ -168,6 +169,17 @@ public void testUpdateNestedColumnPosition() {
     assertEquals(newPosition, updateColumnPosition.getPosition());
   }
 
+  @Test
+  public void testDoesNotAllowDefaultInColumnPosition() {
+    String[] fieldName = {"Name", "Last Name"};
+    ColumnPosition newPosition = TableChange.ColumnPosition.defaultPos();
+    Exception exception =
+        Assertions.assertThrowsExactly(
+            IllegalArgumentException.class,
+            () -> TableChange.updateColumnPosition(fieldName, newPosition));
+    assertEquals("Position cannot be DEFAULT", exception.getMessage());
+  }
+
   @Test
   public void testUpdateColumnComment() {
     String[] fieldName = {"First Name"};

catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/ops/IcebergTableOpsHelper.java

@@ -54,7 +54,6 @@
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.types.Type.PrimitiveType;
 import org.apache.iceberg.types.Types.NestedField;
-import org.apache.iceberg.types.Types.StructType;
 
 public class IcebergTableOpsHelper {
   @VisibleForTesting public static final Joiner DOT = Joiner.on(".");
@@ -133,13 +132,9 @@ private void doMoveColumn(
   }
 
   private void doUpdateColumnPosition(
-      UpdateSchema icebergUpdateSchema,
-      UpdateColumnPosition updateColumnPosition,
-      Schema icebergTableSchema) {
-    StructType tableSchema = icebergTableSchema.asStruct();
-    ColumnPosition columnPosition =
-        getColumnPositionForIceberg(tableSchema, updateColumnPosition.getPosition());
-    doMoveColumn(icebergUpdateSchema, updateColumnPosition.fieldName(), columnPosition);
+      UpdateSchema icebergUpdateSchema, UpdateColumnPosition updateColumnPosition) {
+    doMoveColumn(
+        icebergUpdateSchema, updateColumnPosition.fieldName(), updateColumnPosition.getPosition());
   }
 
   private void doUpdateColumnType(
@@ -160,23 +155,6 @@ private void doUpdateColumnType(
     icebergUpdateSchema.updateColumn(fieldName, (PrimitiveType) type);
   }
 
-  // Iceberg doesn't support LAST position, transform to FIRST or AFTER.
-  private ColumnPosition getColumnPositionForIceberg(
-      StructType parent, ColumnPosition columnPosition) {
-    if (!(columnPosition instanceof TableChange.Default)) {
-      return columnPosition;
-    }
-
-    List<NestedField> fields = parent.fields();
-    // no column, add to first
-    if (fields.isEmpty()) {
-      return ColumnPosition.first();
-    }
-
-    NestedField last = fields.get(fields.size() - 1);
-    return ColumnPosition.after(last.name());
-  }
-
   private void doAddColumn(UpdateSchema icebergUpdateSchema, AddColumn addColumn) {
     if (addColumn.isAutoIncrement()) {
       throw new IllegalArgumentException("Iceberg doesn't support auto increment column");
@@ -230,8 +208,7 @@ private void alterTableColumn(
       } else if (change instanceof DeleteColumn) {
         doDeleteColumn(icebergUpdateSchema, (DeleteColumn) change, icebergTableSchema);
       } else if (change instanceof UpdateColumnPosition) {
-        doUpdateColumnPosition(
-            icebergUpdateSchema, (UpdateColumnPosition) change, icebergTableSchema);
+        doUpdateColumnPosition(icebergUpdateSchema, (UpdateColumnPosition) change);
       } else if (change instanceof RenameColumn) {
         doRenameColumn(icebergUpdateSchema, (RenameColumn) change);
       } else if (change instanceof UpdateColumnType) {

catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/integration/test/CatalogIcebergBaseIT.java

@@ -589,43 +589,6 @@ void testListAndDropIcebergTable() {
     Assertions.assertEquals(0, tableIdentifiers.size());
   }
 
-  @Test
-  public void testUpdateIcebergColumnDefaultPosition() {
-    Column col1 = Column.of("name", Types.StringType.get(), "comment");
-    Column col2 = Column.of("address", Types.StringType.get(), "comment");
-    Column col3 = Column.of("date_of_birth", Types.StringType.get(), "comment");
-    Column[] newColumns = new Column[] {col1, col2, col3};
-    NameIdentifier tableIdentifier =
-        NameIdentifier.of(schemaName, GravitinoITUtils.genRandomName("CatalogIcebergIT_table"));
-    catalog
-        .asTableCatalog()
-        .createTable(
-            tableIdentifier,
-            newColumns,
-            table_comment,
-            ImmutableMap.of(),
-            Transforms.EMPTY_TRANSFORM,
-            Distributions.NONE,
-            new SortOrder[0]);
-
-    catalog
-        .asTableCatalog()
-        .alterTable(
-            tableIdentifier,
-            TableChange.updateColumnPosition(
-                new String[] {col1.name()}, TableChange.ColumnPosition.defaultPos()));
-
-    Table updateColumnPositionTable = catalog.asTableCatalog().loadTable(tableIdentifier);
-
-    Column[] updateCols = updateColumnPositionTable.columns();
-    Assertions.assertEquals(3, updateCols.length);
-    Assertions.assertEquals(col2.name(), updateCols[0].name());
-    Assertions.assertEquals(col3.name(), updateCols[1].name());
-    Assertions.assertEquals(col1.name(), updateCols[2].name());
-
-    catalog.asTableCatalog().dropTable(tableIdentifier);
-  }
-
   @Test
   public void testAlterIcebergTable() {
     Column[] columns = createColumns();