private void collectFieldChanges(Types.NestedField currentField, String parentName, Schema tableSchema,
        List<SchemaChange> changes) {
        String fullName = parentName == null ? currentField.name() : parentName + "." + currentField.name();
        Types.NestedField tableField = tableSchema.findField(fullName);

        if (tableField == null) {
            changes.add(new SchemaChange(SchemaChange.ChangeType.ADD_COLUMN, currentField.name(), currentField.type().asPrimitiveType(), parentName));
        } else {
            // process optional fields
            if (!tableField.isOptional() && currentField.isOptional()) {
                changes.add(new SchemaChange(SchemaChange.ChangeType.MAKE_OPTIONAL, fullName, null, parentName));
            }

            // promote type if needed
            if (!tableField.type().equals(currentField.type()) && canPromoteType(tableField.type(), currentField.type())) {
                changes.add(new SchemaChange(SchemaChange.ChangeType.PROMOTE_TYPE, fullName, currentField.type().asPrimitiveType(), parentName));
            }

            // if it is a nested field, recursively process subfields
            if (currentField.type().isStructType() && tableField.type().isStructType()) {
                List<Types.NestedField> currentSubFields = currentField.type().asStructType().fields();

                for (Types.NestedField currentSubField : currentSubFields) {
                    collectFieldChanges(currentSubField, fullName, tableSchema, changes);
                }
            }
        }
    }
Apply schema changes to the table

@param changes list of schema changes
core/src/main/java/kafka/automq/table/worker/IcebergTableManager.java