fix comments

Author: LiebingYu

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java

@@ -43,7 +43,7 @@
 import java.util.List;
 import java.util.Map;
 
-import static org.apache.fluss.lake.paimon.utils.FlussToPaimonTableChangeConverter.convert;
+import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimonSchemaChanges;
 import static org.apache.fluss.metadata.TableDescriptor.BUCKET_COLUMN_NAME;
 import static org.apache.fluss.metadata.TableDescriptor.OFFSET_COLUMN_NAME;
 import static org.apache.fluss.metadata.TableDescriptor.TIMESTAMP_COLUMN_NAME;
@@ -113,7 +113,7 @@ public void alterTable(TablePath tablePath, List<TableChange> tableChanges)
         try {
             Identifier paimonPath = toPaimonIdentifier(tablePath);
             List<SchemaChange> paimonSchemaChanges =
-                    convert(tableChanges, this::getFlussPropertyKeyToPaimon);
+                    toPaimonSchemaChanges(tableChanges, this::getFlussPropertyKeyToPaimon);
             alterTable(paimonPath, paimonSchemaChanges);
         } catch (Catalog.ColumnAlreadyExistException | Catalog.ColumnNotExistException e) {
             // shouldn't happen before we support schema change

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/FlussToPaimonTableChangeConverter.java

@@ -19,6 +19,7 @@
 
 import org.apache.fluss.lake.paimon.source.FlussRowAsPaimonRow;
 import org.apache.fluss.metadata.ResolvedPartitionSpec;
+import org.apache.fluss.metadata.TableChange;
 import org.apache.fluss.metadata.TablePath;
 import org.apache.fluss.record.ChangeType;
 import org.apache.fluss.row.GenericRow;
@@ -28,13 +29,16 @@
 import org.apache.paimon.data.BinaryRow;
 import org.apache.paimon.data.BinaryRowWriter;
 import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.schema.SchemaChange;
 import org.apache.paimon.types.DataType;
 import org.apache.paimon.types.RowKind;
 import org.apache.paimon.types.RowType;
 
 import javax.annotation.Nullable;
 
+import java.util.ArrayList;
 import java.util.List;
+import java.util.function.Function;
 
 /** Utils for conversion between Paimon and Fluss. */
 public class PaimonConversions {
@@ -106,4 +110,26 @@ public static Object toPaimonLiteral(DataType dataType, Object flussLiteral) {
         return org.apache.paimon.data.InternalRow.createFieldGetter(dataType, 0)
                 .getFieldOrNull(flussRowAsPaimonRow);
     }
+
+    public static List<SchemaChange> toPaimonSchemaChanges(
+            List<TableChange> tableChanges, Function<String, String> optionKeyTransformer) {
+        List<SchemaChange> schemaChanges = new ArrayList<>(tableChanges.size());
+
+        for (TableChange tableChange : tableChanges) {
+            if (tableChange instanceof TableChange.SetOption) {
+                TableChange.SetOption setOption = (TableChange.SetOption) tableChange;
+                schemaChanges.add(
+                        SchemaChange.setOption(
+                                optionKeyTransformer.apply(setOption.getKey()), setOption.getValue()));
+            } else if (tableChange instanceof TableChange.ResetOption) {
+                TableChange.ResetOption resetOption = (TableChange.ResetOption) tableChange;
+                schemaChanges.add(SchemaChange.removeOption(optionKeyTransformer.apply(resetOption.getKey())));
+            } else {
+                throw new UnsupportedOperationException(
+                        "Unsupported table change: " + tableChange.getClass());
+            }
+        }
+
+        return schemaChanges;
+    }
 }