[#1758] feat(autoIncrement) : ColumnChange support auto_increment.

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

@@ -88,7 +88,7 @@ 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);
+    return new AddColumn(fieldName, dataType, null, null, true, false);
   }
 
   /**
@@ -104,7 +104,7 @@ 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);
+    return new AddColumn(fieldName, dataType, comment, null, true, false);
   }
 
   /**
@@ -120,7 +120,7 @@ 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);
+    return new AddColumn(fieldName, dataType, null, position, true, false);
   }
 
   /**
@@ -138,7 +138,7 @@ 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);
+    return new AddColumn(fieldName, dataType, comment, position, true, false);
   }
 
   /**
@@ -154,7 +154,7 @@ 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);
+    return new AddColumn(fieldName, dataType, null, null, nullable, false);
   }
 
   /**
@@ -168,7 +168,7 @@ 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);
+    return new AddColumn(fieldName, dataType, comment, null, nullable, false);
   }
 
   /**
@@ -191,7 +191,32 @@ static TableChange addColumn(
       String comment,
       ColumnPosition position,
       boolean nullable) {
-    return new AddColumn(fieldName, dataType, comment, position, nullable);
+    return new AddColumn(fieldName, dataType, comment, position, nullable, false);
+  }
+
+  /**
+   * 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.
+   * @return A TableChange for the addition.
+   */
+  static TableChange addColumn(
+      String[] fieldName,
+      Type dataType,
+      String comment,
+      ColumnPosition position,
+      boolean nullable,
+      boolean autoIncrement) {
+    return new AddColumn(fieldName, dataType, comment, position, nullable, autoIncrement);
   }
 
   /**
@@ -309,6 +334,17 @@ static TableChange deleteIndex(String name, Boolean ifExists) {
     return new DeleteIndex(name, ifExists);
   }
 
+  /**
+   * Create a TableChange for updating the autoIncrement of a field.
+   *
+   * @param fieldName The field name of the column to update.
+   * @param autoIncrement Whether the column is auto-incremented.
+   * @return A TableChange for the update.
+   */
+  static TableChange updateColumnAutoIncrement(String[] fieldName, boolean autoIncrement) {
+    return new UpdateColumnAutoIncrement(fieldName, autoIncrement);
+  }
+
   /** A TableChange to rename a table. */
   final class RenameTable implements TableChange {
     private final String newName;
@@ -793,17 +829,21 @@ final class AddColumn implements ColumnChange {
     private final ColumnPosition position;
     private final boolean nullable;
 
+    private final boolean autoIncrement;
+
     private AddColumn(
         String[] fieldName,
         Type dataType,
         String comment,
         ColumnPosition position,
-        boolean nullable) {
+        boolean nullable,
+        boolean autoIncrement) {
       this.fieldName = fieldName;
       this.dataType = dataType;
       this.comment = comment;
       this.position = position == null ? ColumnPosition.defaultPos() : position;
       this.nullable = nullable;
+      this.autoIncrement = autoIncrement;
     }
 
     /**
@@ -851,6 +891,15 @@ public boolean isNullable() {
       return nullable;
     }
 
+    /**
+     * Checks if the new column is autoIncrement.
+     *
+     * @return true if the column is autoIncrement; false otherwise.
+     */
+    public boolean isAutoIncrement() {
+      return autoIncrement;
+    }
+
     /**
      * Compares this AddColumn instance with another object for equality. The comparison is based on
      * the field name, data type, comment, position, and nullability.
@@ -864,6 +913,7 @@ public boolean equals(Object o) {
       if (o == null || getClass() != o.getClass()) return false;
       AddColumn addColumn = (AddColumn) o;
       return nullable == addColumn.nullable
+          && autoIncrement == addColumn.autoIncrement
           && Arrays.equals(fieldName, addColumn.fieldName)
           && Objects.equals(dataType, addColumn.dataType)
           && Objects.equals(comment, addColumn.comment)
@@ -872,13 +922,13 @@ public boolean equals(Object o) {
 
     /**
      * Generates a hash code for this AddColumn instance. This hash code is based on the field name,
-     * data type, comment, position, and nullability.
+     * data type, comment, position, nullability, and autoIncrement.
      *
      * @return A hash code value for this column addition operation.
      */
     @Override
     public int hashCode() {
-      int result = Objects.hash(dataType, comment, position, nullable);
+      int result = Objects.hash(dataType, comment, position, nullable, autoIncrement);
       result = 31 * result + Arrays.hashCode(fieldName);
       return result;
     }
@@ -1285,4 +1335,56 @@ public int hashCode() {
       return result;
     }
   }
+
+  /** A TableChange to update the autoIncrement of a field. */
+  final class UpdateColumnAutoIncrement implements ColumnChange {
+    private final String[] fieldName;
+
+    private final boolean autoIncrement;
+
+    /**
+     * Creates a new UpdateColumnAutoIncrement instance.
+     *
+     * @param fieldName The name of the field to be updated.
+     * @param autoIncrement The new autoIncrement flag of the field.
+     */
+    public UpdateColumnAutoIncrement(String[] fieldName, boolean autoIncrement) {
+      this.fieldName = fieldName;
+      this.autoIncrement = autoIncrement;
+    }
+
+    /**
+     * Retrieves the field name of the column whose autoIncrement is being updated.
+     *
+     * @return An array of strings representing the field name.
+     */
+    @Override
+    public String[] fieldName() {
+      return fieldName;
+    }
+
+    /**
+     * The autoIncrement flag of the column.
+     *
+     * @return true if the column is autoIncrement; false otherwise.
+     */
+    public boolean isAutoIncrement() {
+      return autoIncrement;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) return true;
+      if (o == null || getClass() != o.getClass()) return false;
+      UpdateColumnAutoIncrement that = (UpdateColumnAutoIncrement) o;
+      return autoIncrement == that.autoIncrement && Arrays.equals(fieldName, that.fieldName);
+    }
+
+    @Override
+    public int hashCode() {
+      int result = Objects.hash(autoIncrement);
+      result = 31 * result + Arrays.hashCode(fieldName);
+      return result;
+    }
+  }
 }

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

@@ -931,6 +931,9 @@ private void doRemoveProperty(
 
   private void doAddColumn(List<FieldSchema> cols, TableChange.AddColumn change) {
     int targetPosition;
+    if (change.isAutoIncrement()) {
+      throw new IllegalArgumentException("Hive catalog does not support auto-increment column");
+    }
     if (change.getPosition() instanceof TableChange.Default) {
       // add to the end by default
       targetPosition = cols.size();

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

@@ -328,6 +328,11 @@ protected String generateAlterTableSql(
       } else if (change instanceof TableChange.DeleteIndex) {
         lazyLoadTable = getOrCreateTable(databaseName, tableName, lazyLoadTable);
         alterSql.add(deleteIndexDefinition(lazyLoadTable, (TableChange.DeleteIndex) change));
+      } else if (change instanceof TableChange.UpdateColumnAutoIncrement) {
+        lazyLoadTable = getOrCreateTable(databaseName, tableName, lazyLoadTable);
+        alterSql.add(
+            updateColumnAutoIncrementDefinition(
+                lazyLoadTable, (TableChange.UpdateColumnAutoIncrement) change));
       } else {
         throw new IllegalArgumentException(
             "Unsupported table change type: " + change.getClass().getName());
@@ -364,6 +369,29 @@ protected String generateAlterTableSql(
     return result;
   }
 
+  private String updateColumnAutoIncrementDefinition(
+      JdbcTable table, TableChange.UpdateColumnAutoIncrement change) {
+    if (change.fieldName().length > 1) {
+      throw new UnsupportedOperationException("Nested column names are not supported");
+    }
+    String col = change.fieldName()[0];
+    JdbcColumn column = getJdbcColumnFromTable(table, col);
+    JdbcColumn updateColumn =
+        new JdbcColumn.Builder()
+            .withName(col)
+            .withDefaultValue(column.defaultValue())
+            .withNullable(column.nullable())
+            .withType(column.dataType())
+            .withComment(column.comment())
+            .withAutoIncrement(change.isAutoIncrement())
+            .build();
+    return MODIFY_COLUMN
+        + BACK_QUOTE
+        + col
+        + BACK_QUOTE
+        + appendColumnDefinition(updateColumn, new StringBuilder());
+  }
+
   @VisibleForTesting
   static String deleteIndexDefinition(
       JdbcTable lazyLoadTable, TableChange.DeleteIndex deleteIndex) {
@@ -478,6 +506,10 @@ private String addColumnFieldDefinition(TableChange.AddColumn addColumn) {
         .append(dataType)
         .append(SPACE);
 
+    if (addColumn.isAutoIncrement()) {
+      columnDefinition.append(MYSQL_AUTO_INCREMENT).append(SPACE);
+    }
+
     if (!addColumn.isNullable()) {
       columnDefinition.append("NOT NULL ");
     }

catalogs/catalog-jdbc-postgresql/src/main/java/com/datastrato/gravitino/catalog/postgresql/operation/PostgreSqlTableOperations.java

@@ -277,6 +277,10 @@ protected String generateAlterTableSql(
         alterSql.add(addIndexDefinition(tableName, (TableChange.AddIndex) change));
       } else if (change instanceof TableChange.DeleteIndex) {
         alterSql.add(deleteIndexDefinition(tableName, (TableChange.DeleteIndex) change));
+      } else if (change instanceof TableChange.UpdateColumnAutoIncrement) {
+        alterSql.add(
+            updateColumnAutoIncrementDefinition(
+                (TableChange.UpdateColumnAutoIncrement) change, tableName));
       } else {
         throw new IllegalArgumentException(
             "Unsupported table change type: " + change.getClass().getName());
@@ -294,6 +298,21 @@ protected String generateAlterTableSql(
     return result;
   }
 
+  @VisibleForTesting
+  static String updateColumnAutoIncrementDefinition(
+      TableChange.UpdateColumnAutoIncrement change, String tableName) {
+    if (change.fieldName().length > 1) {
+      throw new UnsupportedOperationException(POSTGRESQL_NOT_SUPPORT_NESTED_COLUMN_MSG);
+    }
+    String fieldName = change.fieldName()[0];
+    String action =
+        change.isAutoIncrement() ? "ADD GENERATED BY DEFAULT AS IDENTITY" : "DROP IDENTITY";
+
+    return String.format(
+        "ALTER TABLE %s %s %s %s;",
+        PG_QUOTE + tableName + PG_QUOTE, ALTER_COLUMN, PG_QUOTE + fieldName + PG_QUOTE, action);
+  }
+
   @VisibleForTesting
   static String deleteIndexDefinition(String tableName, TableChange.DeleteIndex deleteIndex) {
     StringBuilder sqlBuilder = new StringBuilder();
@@ -508,6 +527,17 @@ private List<String> addColumnFieldDefinition(
         .append(typeConverter.fromGravitinoType(addColumn.getDataType()))
         .append(SPACE);
 
+    if (addColumn.isAutoIncrement()) {
+      if (!Types.allowAutoIncrement(addColumn.getDataType())) {
+        throw new IllegalArgumentException(
+            "Unsupported auto-increment , column: "
+                + Arrays.toString(addColumn.getFieldName())
+                + ", type: "
+                + addColumn.getDataType());
+      }
+      columnDefinition.append("GENERATED BY DEFAULT AS IDENTITY ");
+    }
+
     // Add NOT NULL if the column is marked as such
     if (!addColumn.isNullable()) {
       columnDefinition.append("NOT NULL ");

catalogs/catalog-jdbc-postgresql/src/test/java/com/datastrato/gravitino/catalog/postgresql/operation/TestPostgreSqlTableOperations.java

@@ -97,4 +97,24 @@ public void testOperationIndexDefinition() {
             + "DROP INDEX IF EXISTS \"test_2_pk\";",
         result);
   }
+
+  @Test
+  void testUpdateColumnAutoIncrementDefinition() {
+    TableChange.UpdateColumnAutoIncrement updateColumnAutoIncrement =
+        new TableChange.UpdateColumnAutoIncrement(new String[] {"col_1"}, true);
+    String sql =
+        PostgreSqlTableOperations.updateColumnAutoIncrementDefinition(
+            updateColumnAutoIncrement, "table_test");
+    Assertions.assertEquals(
+        "ALTER TABLE \"table_test\" ALTER COLUMN  \"col_1\" ADD GENERATED BY DEFAULT AS IDENTITY;",
+        sql);
+
+    updateColumnAutoIncrement =
+        new TableChange.UpdateColumnAutoIncrement(new String[] {"col_2"}, false);
+    sql =
+        PostgreSqlTableOperations.updateColumnAutoIncrementDefinition(
+            updateColumnAutoIncrement, "table_test");
+    Assertions.assertEquals(
+        "ALTER TABLE \"table_test\" ALTER COLUMN  \"col_2\" DROP IDENTITY;", sql);
+  }
 }

catalogs/catalog-lakehouse-iceberg/src/main/java/com/datastrato/gravitino/catalog/lakehouse/iceberg/ops/IcebergTableOpsHelper.java

@@ -191,6 +191,10 @@ private void doAddColumn(
       parentStruct = icebergTableSchema.asStruct();
     }
 
+    if (addColumn.isAutoIncrement()) {
+      throw new IllegalArgumentException("Iceberg doesn't support auto increment column");
+    }
+
     if (addColumn.isNullable()) {
       icebergUpdateSchema.addColumn(
           getParentName(addColumn.fieldName()),

clients/client-java/src/main/java/com/datastrato/gravitino/client/DTOConverters.java

@@ -190,7 +190,8 @@ private static TableUpdateRequest toColumnUpdateRequest(TableChange.ColumnChange
           addColumn.getDataType(),
           addColumn.getComment(),
           addColumn.getPosition(),
-          addColumn.isNullable());
+          addColumn.isNullable(),
+          addColumn.isAutoIncrement());
 
     } else if (change instanceof TableChange.RenameColumn) {
       TableChange.RenameColumn renameColumn = (TableChange.RenameColumn) change;
@@ -216,6 +217,9 @@ private static TableUpdateRequest toColumnUpdateRequest(TableChange.ColumnChange
     } else if (change instanceof TableChange.UpdateColumnNullability) {
       return new TableUpdateRequest.UpdateTableColumnNullabilityRequest(
           change.fieldName(), ((TableChange.UpdateColumnNullability) change).nullable());
+    } else if (change instanceof TableChange.UpdateColumnAutoIncrement) {
+      return new TableUpdateRequest.UpdateColumnAutoIncrementRequest(
+          change.fieldName(), ((TableChange.UpdateColumnAutoIncrement) change).isAutoIncrement());
     } else {
       throw new IllegalArgumentException(
           "Unknown column change type: " + change.getClass().getSimpleName());