[#839] feat: AddColumn supports adding default values (#2558)

### What changes were proposed in this pull request?
Supports setting default value when adding column.
- MySQL & PostgreSQL are supported
- Hive & Iceberg are not supported -> throws exception

### Why are the changes needed?
Fix: #839

### Does this PR introduce _any_ user-facing change?
Users are able to set default value for column when adding column with
`com.datastrato.gravitino.rel.TableCatalog#alterTable`

### How was this patch tested?
Add tests of adding columns with default value in existing tests.

api/src/main/java/com/datastrato/gravitino/rel/TableChange.java

@@ -21,6 +21,7 @@
 package com.datastrato.gravitino.rel;
 
 import com.datastrato.gravitino.annotation.Evolving;
+import com.datastrato.gravitino.rel.expressions.Expression;
 import com.datastrato.gravitino.rel.indexes.Index;
 import com.datastrato.gravitino.rel.types.Type;
 import java.util.Arrays;
@@ -32,7 +33,6 @@
  */
 @Evolving
 public interface TableChange {
-
   /**
    * Create a TableChange for renaming a table.
    *
@@ -90,7 +90,8 @@ static TableChange removeProperty(String property) {
    * @return A TableChange for the addition.
    */
   static TableChange addColumn(String[] fieldName, Type dataType) {
-    return new AddColumn(fieldName, dataType, null, null, true, false);
+    return new AddColumn(
+        fieldName, dataType, null, null, true, false, Column.DEFAULT_VALUE_NOT_SET);
   }
 
   /**
@@ -106,7 +107,8 @@ static TableChange addColumn(String[] fieldName, Type dataType) {
    * @return A TableChange for the addition.
    */
   static TableChange addColumn(String[] fieldName, Type dataType, String comment) {
-    return new AddColumn(fieldName, dataType, comment, null, true, false);
+    return new AddColumn(
+        fieldName, dataType, comment, null, true, false, Column.DEFAULT_VALUE_NOT_SET);
   }
 
   /**
@@ -122,7 +124,8 @@ static TableChange addColumn(String[] fieldName, Type dataType, String comment)
    * @return A TableChange for the addition.
    */
   static TableChange addColumn(String[] fieldName, Type dataType, ColumnPosition position) {
-    return new AddColumn(fieldName, dataType, null, position, true, false);
+    return new AddColumn(
+        fieldName, dataType, null, position, true, false, Column.DEFAULT_VALUE_NOT_SET);
   }
 
   /**
@@ -140,7 +143,8 @@ static TableChange addColumn(String[] fieldName, Type dataType, ColumnPosition p
    */
   static TableChange addColumn(
       String[] fieldName, Type dataType, String comment, ColumnPosition position) {
-    return new AddColumn(fieldName, dataType, comment, position, true, false);
+    return new AddColumn(
+        fieldName, dataType, comment, position, true, false, Column.DEFAULT_VALUE_NOT_SET);
   }
 
   /**
@@ -156,7 +160,8 @@ static TableChange addColumn(
    * @return A TableChange for the addition.
    */
   static TableChange addColumn(String[] fieldName, Type dataType, boolean nullable) {
-    return new AddColumn(fieldName, dataType, null, null, nullable, false);
+    return new AddColumn(
+        fieldName, dataType, null, null, nullable, false, Column.DEFAULT_VALUE_NOT_SET);
   }
 
   /**
@@ -170,7 +175,67 @@ static TableChange addColumn(String[] fieldName, Type dataType, boolean nullable
    */
   static TableChange addColumn(
       String[] fieldName, Type dataType, String comment, boolean nullable) {
-    return new AddColumn(fieldName, dataType, comment, null, nullable, false);
+    return new AddColumn(
+        fieldName, dataType, comment, null, nullable, false, Column.DEFAULT_VALUE_NOT_SET);
+  }
+
+  /**
+   * Create a TableChange for adding a column.
+   *
+   * @param fieldName Field name of the new column.
+   * @param dataType The new column's data type.
+   * @param defaultValue The new column's default value.
+   * @return A TableChange for the addition.
+   */
+  static TableChange addColumn(String[] fieldName, Type dataType, Expression defaultValue) {
+    return new AddColumn(fieldName, dataType, null, null, true, false, defaultValue);
+  }
+
+  /**
+   * Create a TableChange for adding a column.
+   *
+   * @param fieldName Field name of the new column.
+   * @param dataType The new column's data type.
+   * @param comment The new field's comment string.
+   * @param defaultValue The new column's default value.
+   * @return A TableChange for the addition.
+   */
+  static TableChange addColumn(
+      String[] fieldName, Type dataType, String comment, Expression defaultValue) {
+    return new AddColumn(fieldName, dataType, comment, null, true, false, defaultValue);
+  }
+
+  /**
+   * Create a TableChange for adding a column.
+   *
+   * @param fieldName Field name of the new column.
+   * @param dataType The new column's data type.
+   * @param position The new column's position.
+   * @param defaultValue The new column's default value.
+   * @return A TableChange for the addition.
+   */
+  static TableChange addColumn(
+      String[] fieldName, Type dataType, ColumnPosition position, Expression defaultValue) {
+    return new AddColumn(fieldName, dataType, null, position, true, false, defaultValue);
+  }
+
+  /**
+   * Create a TableChange for adding a column.
+   *
+   * @param fieldName Field name of the new column.
+   * @param dataType The new column's data type.
+   * @param comment The new field's comment string.
+   * @param position The new column's position.
+   * @param defaultValue The new column's default value.
+   * @return A TableChange for the addition.
+   */
+  static TableChange addColumn(
+      String[] fieldName,
+      Type dataType,
+      String comment,
+      ColumnPosition position,
+      Expression defaultValue) {
+    return new AddColumn(fieldName, dataType, comment, position, true, false, defaultValue);
   }
 
   /**
@@ -193,7 +258,8 @@ static TableChange addColumn(
       String comment,
       ColumnPosition position,
       boolean nullable) {
-    return new AddColumn(fieldName, dataType, comment, position, nullable, false);
+    return new AddColumn(
+        fieldName, dataType, comment, position, nullable, false, Column.DEFAULT_VALUE_NOT_SET);
   }
 
   /**
@@ -218,7 +284,42 @@ static TableChange addColumn(
       ColumnPosition position,
       boolean nullable,
       boolean autoIncrement) {
-    return new AddColumn(fieldName, dataType, comment, position, nullable, autoIncrement);
+    return new AddColumn(
+        fieldName,
+        dataType,
+        comment,
+        position,
+        nullable,
+        autoIncrement,
+        Column.DEFAULT_VALUE_NOT_SET);
+  }
+
+  /**
+   * Create a TableChange for adding a column.
+   *
+   * <p>If the field already exists, the change will result in an {@link IllegalArgumentException}.
+   * If the new field is nested and its parent does not exist or is not a struct, the change will
+   * result in an {@link IllegalArgumentException}.
+   *
+   * @param fieldName Field name of the new column.
+   * @param dataType The new column's data type.
+   * @param comment The new field's comment string.
+   * @param position The new column's position.
+   * @param nullable The new column's nullable.
+   * @param autoIncrement The new column's autoIncrement.
+   * @param defaultValue The new column's default value.
+   * @return A TableChange for the addition.
+   */
+  static TableChange addColumn(
+      String[] fieldName,
+      Type dataType,
+      String comment,
+      ColumnPosition position,
+      boolean nullable,
+      boolean autoIncrement,
+      Expression defaultValue) {
+    return new AddColumn(
+        fieldName, dataType, comment, position, nullable, autoIncrement, defaultValue);
   }
 
   /**
@@ -830,22 +931,24 @@ final class AddColumn implements ColumnChange {
     private final String comment;
     private final ColumnPosition position;
     private final boolean nullable;
-
     private final boolean autoIncrement;
+    private final Expression defaultValue;
 
     private AddColumn(
         String[] fieldName,
         Type dataType,
         String comment,
         ColumnPosition position,
         boolean nullable,
-        boolean autoIncrement) {
+        boolean autoIncrement,
+        Expression defaultValue) {
       this.fieldName = fieldName;
       this.dataType = dataType;
       this.comment = comment;
       this.position = position == null ? ColumnPosition.defaultPos() : position;
       this.nullable = nullable;
       this.autoIncrement = autoIncrement;
+      this.defaultValue = defaultValue;
     }
 
     /**
@@ -902,6 +1005,15 @@ public boolean isAutoIncrement() {
       return autoIncrement;
     }
 
+    /**
+     * Retrieves the default value of the new column.
+     *
+     * @return The default value of the column.
+     */
+    public Expression getDefaultValue() {
+      return defaultValue;
+    }
+
     /**
      * Compares this AddColumn instance with another object for equality. The comparison is based on
      * the field name, data type, comment, position, and nullability.
@@ -919,7 +1031,8 @@ public boolean equals(Object o) {
           && Arrays.equals(fieldName, addColumn.fieldName)
           && Objects.equals(dataType, addColumn.dataType)
           && Objects.equals(comment, addColumn.comment)
-          && Objects.equals(position, addColumn.position);
+          && Objects.equals(position, addColumn.position)
+          && Objects.equals(defaultValue, addColumn.defaultValue);
     }
 
     /**
@@ -930,7 +1043,7 @@ public boolean equals(Object o) {
      */
     @Override
     public int hashCode() {
-      int result = Objects.hash(dataType, comment, position, nullable, autoIncrement);
+      int result = Objects.hash(dataType, comment, position, nullable, autoIncrement, defaultValue);
       result = 31 * result + Arrays.hashCode(fieldName);
       return result;
     }

api/src/test/java/com/datastrato/gravitino/TestTableChange.java

@@ -11,6 +11,7 @@
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import com.datastrato.gravitino.rel.Column;
 import com.datastrato.gravitino.rel.TableChange;
 import com.datastrato.gravitino.rel.TableChange.AddColumn;
 import com.datastrato.gravitino.rel.TableChange.ColumnPosition;
@@ -76,6 +77,7 @@ public void testAddColumn() {
     assertEquals(dataType, addColumn.getDataType());
     assertEquals(comment, addColumn.getComment());
     assertEquals(ColumnPosition.defaultPos(), addColumn.getPosition());
+    assertEquals(Column.DEFAULT_VALUE_NOT_SET, addColumn.getDefaultValue());
   }
 
   @Test
@@ -90,18 +92,21 @@ public void testAddColumnWithPosition() {
     assertEquals(dataType, addColumn.getDataType());
     assertEquals(comment, addColumn.getComment());
     assertEquals(position, addColumn.getPosition());
+    assertEquals(Column.DEFAULT_VALUE_NOT_SET, addColumn.getDefaultValue());
   }
 
   @Test
   public void testAddColumnWithNullCommentAndPosition() {
     String[] fieldName = {"Middle Name"};
     Type dataType = Types.StringType.get();
-    AddColumn addColumn = (AddColumn) TableChange.addColumn(fieldName, dataType, null, null);
+    AddColumn addColumn =
+        (AddColumn) TableChange.addColumn(fieldName, dataType, null, (ColumnPosition) null);
 
     assertArrayEquals(fieldName, addColumn.fieldName());
     assertEquals(dataType, addColumn.getDataType());
     assertNull(addColumn.getComment());
     assertEquals(ColumnPosition.defaultPos(), addColumn.getPosition());
+    assertEquals(Column.DEFAULT_VALUE_NOT_SET, addColumn.getDefaultValue());
   }
 
   @Test

catalogs/catalog-hive/src/main/java/com/datastrato/gravitino/catalog/hive/HiveCatalogOperations.java

@@ -785,7 +785,9 @@ public Table alterTable(NameIdentifier tableIdent, TableChange... changes)
 
           if (change instanceof TableChange.AddColumn) {
             TableChange.AddColumn addColumn = (TableChange.AddColumn) change;
-            validateNullable(String.join(".", addColumn.fieldName()), addColumn.isNullable());
+            String fieldName = String.join(".", addColumn.fieldName());
+            validateNullable(fieldName, addColumn.isNullable());
+            validateColumnDefaultValue(fieldName, addColumn.getDefaultValue());
             doAddColumn(cols, addColumn);
 
           } else if (change instanceof TableChange.DeleteColumn) {

catalogs/catalog-hive/src/test/java/com/datastrato/gravitino/catalog/hive/TestHiveTable.java

@@ -497,6 +497,21 @@ public void testAlterHiveTable() {
     Assertions.assertEquals(
         "Hive does not support altering column nullability", exception.getMessage());
 
+    TableChange tableChange8 =
+        TableChange.addColumn(
+            new String[] {"col_3"}, Types.ByteType.get(), "comment", Literals.NULL);
+    exception =
+        Assertions.assertThrows(
+            IllegalArgumentException.class,
+            () -> tableCatalog.alterTable(tableIdentifier, tableChange8));
+    Assertions.assertTrue(
+        exception
+            .getMessage()
+            .contains(
+                "The DEFAULT constraint for column is only supported since Hive 3.0, "
+                    + "but the current Gravitino Hive catalog only supports Hive 2.x"),
+        "The exception message is: " + exception.getMessage());
+
     // test alter
     tableCatalog.alterTable(
         tableIdentifier,

catalogs/catalog-hive/src/test/java/com/datastrato/gravitino/catalog/hive/integration/test/CatalogHiveIT.java

@@ -971,6 +971,21 @@ public void testAlterHiveTable() throws TException, InterruptedException {
             });
     Assertions.assertTrue(exception.getMessage().contains("Cannot alter partition column"));
 
+    // test add column with default value exception
+    TableChange withDefaultValue =
+        TableChange.addColumn(
+            new String[] {"col_3"}, Types.ByteType.get(), "comment", Literals.NULL);
+    exception =
+        Assertions.assertThrows(
+            IllegalArgumentException.class, () -> tableCatalog.alterTable(id, withDefaultValue));
+    Assertions.assertTrue(
+        exception
+            .getMessage()
+            .contains(
+                "The DEFAULT constraint for column is only supported since Hive 3.0, "
+                    + "but the current Gravitino Hive catalog only supports Hive 2.x"),
+        "The exception message is: " + exception.getMessage());
+
     // test updateColumnPosition exception
     Column col1 = Column.of("name", Types.StringType.get(), "comment");
     Column col2 = Column.of("address", Types.StringType.get(), "comment");

catalogs/catalog-jdbc-mysql/src/main/java/com/datastrato/gravitino/catalog/mysql/operation/MysqlTableOperations.java

@@ -534,6 +534,14 @@ private String addColumnFieldDefinition(TableChange.AddColumn addColumn) {
       columnDefinition.append("COMMENT '").append(addColumn.getComment()).append("' ");
     }
 
+    // Append default value if available
+    if (!Column.DEFAULT_VALUE_NOT_SET.equals(addColumn.getDefaultValue())) {
+      columnDefinition
+          .append("DEFAULT ")
+          .append(columnDefaultValueConverter.fromGravitino(addColumn.getDefaultValue()))
+          .append(SPACE);
+    }
+
     // Append position if available
     if (addColumn.getPosition() instanceof TableChange.First) {
       columnDefinition.append("FIRST");