[lake/paimon] Fix schema evolution safety and address review comments

- Add shouldAlterTable() to verify schema before altering
- Add isColumnAlreadyExists() for precise column property validation
- Fix null-safe comment comparison logic
- Use containsExactly() for precise field order assertions in tests

This addresses all review comments from wuchong in PR apache#2189

Author: buvb

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

@@ -109,15 +109,87 @@ public void alterTable(TablePath tablePath, List<TableChange> tableChanges, Cont
             throws TableNotExistException {
         try {
             List<SchemaChange> paimonSchemaChanges = toPaimonSchemaChanges(tableChanges);
-            alterTable(tablePath, paimonSchemaChanges);
+
+            // Compare current Paimon table schema with expected target schema before altering
+            if (shouldAlterTable(tablePath, tableChanges)) {
+                alterTable(tablePath, paimonSchemaChanges);
+            }
+            // If schemas already match, treat as idempotent success
         } catch (Catalog.ColumnAlreadyExistException e) {
-            // Column already exists, treat as idempotent success for retry scenarios.
+            // This shouldn't happen if shouldAlterTable works correctly, but keep as safeguard
+            throw new RuntimeException("Unexpected ColumnAlreadyExistException", e);
         } catch (Catalog.ColumnNotExistException e) {
             // This shouldn't happen for AddColumn operations
             throw new RuntimeException(e);
         }
     }
 
+    private boolean shouldAlterTable(TablePath tablePath, List<TableChange> tableChanges)
+            throws TableNotExistException {
+        try {
+            Table table = paimonCatalog.getTable(toPaimon(tablePath));
+            FileStoreTable fileStoreTable = (FileStoreTable) table;
+            Schema currentSchema = fileStoreTable.schema().toSchema();
+
+            for (TableChange change : tableChanges) {
+                if (change instanceof TableChange.AddColumn) {
+                    TableChange.AddColumn addColumn = (TableChange.AddColumn) change;
+                    if (!isColumnAlreadyExists(currentSchema, addColumn)) {
+                        return true;
+                    }
+                } else {
+                    return true;
+                }
+            }
+
+            return false;
+        } catch (Catalog.TableNotExistException e) {
+            throw new TableNotExistException("Table " + tablePath + " does not exist.");
+        }
+    }
+
+    private boolean isColumnAlreadyExists(Schema currentSchema, TableChange.AddColumn addColumn) {
+        String columnName = addColumn.getName();
+
+        for (org.apache.paimon.types.DataField field : currentSchema.fields()) {
+            if (field.name().equals(columnName)) {
+                org.apache.paimon.types.DataType expectedType =
+                        addColumn
+                                .getDataType()
+                                .accept(
+                                        org.apache.fluss.lake.paimon.utils
+                                                .FlussDataTypeToPaimonDataType.INSTANCE);
+
+                if (!field.type().equals(expectedType)) {
+                    throw new IllegalStateException(
+                            String.format(
+                                    "Column %s already exists but with different type. "
+                                            + "Existing: %s, Expected: %s",
+                                    columnName, field.type(), expectedType));
+                }
+                String existingComment = field.description();
+                String expectedComment = addColumn.getComment();
+
+                boolean commentsMatch =
+                        (existingComment == null && expectedComment == null)
+                                || (existingComment != null
+                                        && existingComment.equals(expectedComment));
+
+                if (!commentsMatch) {
+                    throw new IllegalStateException(
+                            String.format(
+                                    "Column %s already exists but with different comment. "
+                                            + "Existing: %s, Expected: %s",
+                                    columnName, existingComment, expectedComment));
+                }
+
+                return true;
+            }
+        }
+
+        return false;
+    }
+
     private void createTable(TablePath tablePath, Schema schema, boolean isCreatingFlussTable)
             throws Catalog.DatabaseNotExistException {
         Identifier paimonPath = toPaimon(tablePath);

fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/FlussRecordAsPaimonRowTest.java

@@ -40,6 +40,7 @@
 import static org.apache.fluss.record.ChangeType.UPDATE_AFTER;
 import static org.apache.fluss.record.ChangeType.UPDATE_BEFORE;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** Test for {@link FlussRecordAsPaimonRow}. */
 class FlussRecordAsPaimonRowTest {
@@ -597,13 +598,11 @@ void testFlussRecordWiderThanPaimonSchema() {
         genericRow.setField(0, true);
         genericRow.setField(1, BinaryString.fromString("extra"));
         LogRecord logRecord = new GenericRecord(logOffset, timeStamp, APPEND_ONLY, genericRow);
-        flussRecordAsPaimonRow.setFlussRecord(logRecord);
-
-        assertThat(flussRecordAsPaimonRow.getFieldCount()).isEqualTo(4);
-        assertThat(flussRecordAsPaimonRow.getBoolean(0)).isTrue();
 
-        assertThat(flussRecordAsPaimonRow.getInt(1)).isEqualTo(tableBucket);
-        assertThat(flussRecordAsPaimonRow.getLong(2)).isEqualTo(logOffset);
-        assertThat(flussRecordAsPaimonRow.getLong(3)).isEqualTo(timeStamp);
+        // Should throw exception instead of silently truncating data
+        assertThatThrownBy(() -> flussRecordAsPaimonRow.setFlussRecord(logRecord))
+                .isInstanceOf(IllegalStateException.class)
+                .hasMessageContaining(
+                        "Fluss record has 2 fields but Paimon schema only has 1 business fields");
     }
 }

fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringITCase.java

@@ -643,20 +643,30 @@ void testTieringWithAddColumn() throws Exception {
             // 7. Wait for new data to be tiered
             assertReplicaStatus(tableBucket, 6);
 
-            // 8. Verify Paimon table has the new column
+            // 8. Verify Paimon table has the new column with exact field names and order
             Identifier tableIdentifier =
                     Identifier.create(tablePath.getDatabaseName(), tablePath.getTableName());
             FileStoreTable paimonTable = (FileStoreTable) paimonCatalog.getTable(tableIdentifier);
             List<String> fieldNames = paimonTable.rowType().getFieldNames();
 
-            // Should have: a, b, c3, __bucket, __offset, __timestamp
-            assertThat(fieldNames).contains("a", "b", "c3");
-
-            // 9. Verify all data is present in Paimon (no data loss)
-            List<InternalRow> allRows = new ArrayList<>();
-            allRows.addAll(initialRows);
-            allRows.addAll(newRows);
-            checkDataInPaimonAppendOnlyTable(tablePath, allRows, 0);
+            // Should have exact fields in order: a, b, c3, __bucket, __offset, __timestamp
+            assertThat(fieldNames)
+                    .containsExactly("a", "b", "c3", "__bucket", "__offset", "__timestamp");
+
+            // 9. Verify both schema evolution and data correctness
+            // For initial rows (before ADD COLUMN), c3 should be NULL
+            // For new rows (after ADD COLUMN), c3 should have the provided values
+            List<InternalRow> expectedRows = new ArrayList<>();
+            // Initial rows with NULL for c3
+            expectedRows.add(row(1, "v1", null));
+            expectedRows.add(row(2, "v2", null));
+            expectedRows.add(row(3, "v3", null));
+            // New rows with c3 values
+            expectedRows.add(row(4, "v4", 40));
+            expectedRows.add(row(5, "v5", 50));
+            expectedRows.add(row(6, "v6", 60));
+
+            checkDataInPaimonAppendOnlyTable(tablePath, expectedRows, 0);
 
         } finally {
             jobClient.cancel().get();