[paimon] Align error msg when altering table properties

Author: nastra

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

@@ -109,9 +109,8 @@ public void createTable(TablePath tablePath, TableDescriptor tableDescriptor, Co
     public void alterTable(TablePath tablePath, List<TableChange> tableChanges, Context context)
             throws TableNotExistException {
         try {
-            Identifier paimonPath = toPaimon(tablePath);
             List<SchemaChange> paimonSchemaChanges = toPaimonSchemaChanges(tableChanges);
-            alterTable(paimonPath, paimonSchemaChanges);
+            alterTable(tablePath, paimonSchemaChanges);
         } catch (Catalog.ColumnAlreadyExistException | Catalog.ColumnNotExistException e) {
             // shouldn't happen before we support schema change
             throw new RuntimeException(e);
@@ -155,12 +154,12 @@ private void createDatabase(String databaseName) {
         }
     }
 
-    private void alterTable(Identifier tablePath, List<SchemaChange> tableChanges)
+    private void alterTable(TablePath tablePath, List<SchemaChange> tableChanges)
             throws Catalog.ColumnAlreadyExistException, Catalog.ColumnNotExistException {
         try {
-            paimonCatalog.alterTable(tablePath, tableChanges, false);
+            paimonCatalog.alterTable(toPaimon(tablePath), tableChanges, false);
         } catch (Catalog.TableNotExistException e) {
-            throw new TableNotExistException("Table " + tablePath + " not exists.");
+            throw new TableNotExistException("Table " + tablePath + " does not exist.");
         }
     }
 

fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/PaimonLakeCatalogTest.java

@@ -18,6 +18,7 @@
 package org.apache.fluss.lake.paimon;
 
 import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.TableNotExistException;
 import org.apache.fluss.lake.lakestorage.TestingLakeCatalogContext;
 import org.apache.fluss.metadata.Schema;
 import org.apache.fluss.metadata.TableChange;
@@ -35,6 +36,7 @@
 import java.util.Collections;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** Unit test for {@link PaimonLakeCatalog}. */
 class PaimonLakeCatalogTest {
@@ -83,6 +85,34 @@ void testAlterTableProperties() throws Exception {
         assertThat(table.options().get("fluss.key")).isEqualTo(null);
     }
 
+    @Test
+    void alterTablePropertiesWithNonExistentTable() {
+        TestingLakeCatalogContext context = new TestingLakeCatalogContext();
+        // db & table don't exist
+        assertThatThrownBy(
+                        () ->
+                                flussPaimonCatalog.alterTable(
+                                        TablePath.of("non_existing_db", "non_existing_table"),
+                                        Collections.singletonList(TableChange.set("key", "value")),
+                                        context))
+                .isInstanceOf(TableNotExistException.class)
+                .hasMessage("Table non_existing_db.non_existing_table does not exist.");
+
+        String database = "alter_props_db";
+        String tableName = "alter_props_table";
+        createTable(database, tableName);
+
+        // database exists but table doesn't
+        assertThatThrownBy(
+                        () ->
+                                flussPaimonCatalog.alterTable(
+                                        TablePath.of(database, "non_existing_table"),
+                                        Collections.singletonList(TableChange.set("key", "value")),
+                                        context))
+                .isInstanceOf(TableNotExistException.class)
+                .hasMessage("Table alter_props_db.non_existing_table does not exist.");
+    }
+
     private void createTable(String database, String tableName) {
         Schema flussSchema =
                 Schema.newBuilder()