[feature](paimon) Support Paimon DDL and external table properties

fe/fe-core/src/main/java/org/apache/doris/alter/Alter.java

@@ -396,7 +396,13 @@ private void processAlterTableForExternalTable(
         long updateTime = System.currentTimeMillis();
         for (AlterOp alterOp : alterOps) {
             if (alterOp instanceof ModifyTablePropertiesOp) {
-                setExternalTableAutoAnalyzePolicy(table, alterOps);
+                Map<String, String> properties = alterOp.getProperties();
+                if (properties.size() == 1
+                        && properties.containsKey(PropertyAnalyzer.PROPERTIES_AUTO_ANALYZE_POLICY)) {
+                    setExternalTableAutoAnalyzePolicy(table, alterOps);
+                } else {
+                    table.getCatalog().updateTableProperties(table, properties);
+                }
             } else if (alterOp instanceof CreateOrReplaceBranchOp) {
                 table.getCatalog().createOrReplaceBranch(
                         table, ((CreateOrReplaceBranchOp) alterOp).getBranchInfo());

fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java

@@ -1597,6 +1597,25 @@ public void renameColumn(TableIf dorisTable, String oldName, String newName) thr
         }
     }
 
+    public void updateTableProperties(TableIf dorisTable, Map<String, String> properties) throws UserException {
+        makeSureInitialized();
+        Preconditions.checkState(dorisTable instanceof ExternalTable, dorisTable.getName());
+        ExternalTable externalTable = (ExternalTable) dorisTable;
+        if (metadataOps == null) {
+            throw new DdlException("Update table properties operation is not supported for catalog: " + getName());
+        }
+        try {
+            long updateTime = System.currentTimeMillis();
+            metadataOps.updateTableProperties(externalTable, properties);
+            Env.getCurrentEnv().getExtMetaCacheMgr().invalidateTableCache(externalTable);
+            logRefreshExternalTable(externalTable, updateTime);
+        } catch (Exception e) {
+            LOG.warn("Failed to update properties for table {}.{} in catalog {}",
+                    externalTable.getDbName(), externalTable.getName(), getName(), e);
+            throw e;
+        }
+    }
+
     @Override
     public void modifyColumn(TableIf dorisTable, Column column, ColumnPosition columnPosition) throws UserException {
         makeSureInitialized();

fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergMetadataOps.java

@@ -68,6 +68,7 @@
 import org.apache.iceberg.Table;
 import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.UpdatePartitionSpec;
+import org.apache.iceberg.UpdateProperties;
 import org.apache.iceberg.UpdateSchema;
 import org.apache.iceberg.catalog.Catalog;
 import org.apache.iceberg.catalog.Namespace;
@@ -848,6 +849,20 @@ public void renameColumn(ExternalTable dorisTable, String oldName, String newNam
         refreshTable(dorisTable, updateTime);
     }
 
+    @Override
+    public void updateTableProperties(ExternalTable dorisTable, Map<String, String> properties)
+            throws UserException {
+        Table icebergTable = IcebergUtils.getIcebergTable(dorisTable);
+        UpdateProperties updateProperties = icebergTable.updateProperties();
+        properties.forEach(updateProperties::set);
+        try {
+            executionAuthenticator.execute(() -> updateProperties.commit());
+        } catch (Exception e) {
+            throw new UserException("Failed to update properties for table: " + icebergTable.name()
+                    + ", error message is: " + e.getMessage(), e);
+        }
+    }
+
     @Override
     public void modifyColumn(ExternalTable dorisTable, Column column, ColumnPosition position, long updateTime)
             throws UserException {

fe/fe-core/src/main/java/org/apache/doris/datasource/operations/ExternalMetadataOps.java

@@ -263,6 +263,19 @@ default void renameColumn(ExternalTable dorisTable, String oldName, String newNa
         throw new UnsupportedOperationException("Rename column operation is not supported for this table type.");
     }
 
+    /**
+     * update properties for external table
+     *
+     * @param dorisTable external table
+     * @param properties properties to update
+     * @throws UserException if the update fails
+     */
+    default void updateTableProperties(ExternalTable dorisTable, Map<String, String> properties)
+            throws UserException {
+        throw new UnsupportedOperationException(
+                "Update table properties operation is not supported for this table type.");
+    }
+
     /**
      * update column for external table
      *

fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonMetadataOps.java

@@ -18,9 +18,12 @@
 package org.apache.doris.datasource.paimon;
 
 import org.apache.doris.analysis.PartitionDesc;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.Env;
 import org.apache.doris.catalog.StructField;
 import org.apache.doris.catalog.StructType;
 import org.apache.doris.catalog.Type;
+import org.apache.doris.catalog.info.ColumnPosition;
 import org.apache.doris.catalog.info.CreateOrReplaceBranchInfo;
 import org.apache.doris.catalog.info.CreateOrReplaceTagInfo;
 import org.apache.doris.catalog.info.DropBranchInfo;
@@ -50,6 +53,7 @@
 import org.apache.paimon.catalog.Catalog.TableNotExistException;
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaChange;
 import org.apache.paimon.types.DataType;
 
 import java.util.ArrayList;
@@ -269,6 +273,30 @@ public void afterCreateTable(String dbName, String tblName) {
                 dorisCatalog.getName(), dbName, tblName, db.isPresent());
     }
 
+    @Override
+    public void renameTableImpl(String dbName, String oldName, String newName) throws DdlException {
+        try {
+            executionAuthenticator.execute(() -> {
+                catalog.renameTable(Identifier.create(dbName, oldName), Identifier.create(dbName, newName), false);
+                return null;
+            });
+        } catch (Exception e) {
+            throw new DdlException("Failed to rename Paimon table " + dbName + "." + oldName
+                    + " to " + newName + ": " + e.getMessage(), e);
+        }
+    }
+
+    @Override
+    public void afterRenameTable(String dbName, String oldName, String newName) {
+        Optional<ExternalDatabase<?>> db = dorisCatalog.getDbForReplay(dbName);
+        if (db.isPresent()) {
+            db.get().unregisterTable(oldName);
+            db.get().resetMetaCacheNames();
+        }
+        LOG.info("after rename table {}.{}.{} to {}, is db exists: {}",
+                dorisCatalog.getName(), dbName, oldName, newName, db.isPresent());
+    }
+
     @Override
     public void dropTableImpl(ExternalTable dorisTable, boolean ifExists) throws DdlException {
         try {
@@ -392,6 +420,83 @@ public boolean databaseExist(String dbName) {
         }
     }
 
+    private SchemaChange addColumnChange(Column column, ColumnPosition position) {
+        DataType paimonType = toPaimontype(column.getType()).copy(column.isAllowNull());
+        SchemaChange.Move move = null;
+        if (position != null) {
+            move = position.isFirst()
+                    ? SchemaChange.Move.first(column.getName())
+                    : SchemaChange.Move.after(column.getName(), position.getLastCol());
+        }
+        return SchemaChange.addColumn(column.getName(), paimonType, column.getComment(), move);
+    }
+
+    private void alterTable(ExternalTable dorisTable, List<SchemaChange> changes, String operation)
+            throws UserException {
+        try {
+            executionAuthenticator.execute(() -> {
+                catalog.alterTable(
+                        Identifier.create(dorisTable.getRemoteDbName(), dorisTable.getRemoteName()), changes, false);
+                return null;
+            });
+        } catch (Exception e) {
+            throw new UserException("Failed to " + operation + " for Paimon table " + dorisTable.getName()
+                    + ": " + e.getMessage(), e);
+        }
+    }
+
+    private void refreshTable(ExternalTable dorisTable, long updateTime) {
+        Optional<ExternalDatabase<?>> db = dorisCatalog.getDbForReplay(dorisTable.getRemoteDbName());
+        if (db.isPresent()) {
+            Optional<?> table = db.get().getTableForReplay(dorisTable.getRemoteName());
+            if (table.isPresent()) {
+                Env.getCurrentEnv().getRefreshManager()
+                        .refreshTableInternal(db.get(), (ExternalTable) table.get(), updateTime);
+            }
+        }
+    }
+
+    @Override
+    public void addColumn(ExternalTable dorisTable, Column column, ColumnPosition position, long updateTime)
+            throws UserException {
+        alterTable(dorisTable, Collections.singletonList(addColumnChange(column, position)),
+                "add column " + column.getName());
+        refreshTable(dorisTable, updateTime);
+    }
+
+    @Override
+    public void addColumns(ExternalTable dorisTable, List<Column> columns, long updateTime) throws UserException {
+        List<SchemaChange> changes = columns.stream()
+                .map(column -> addColumnChange(column, null))
+                .collect(Collectors.toList());
+        alterTable(dorisTable, changes, "add columns");
+        refreshTable(dorisTable, updateTime);
+    }
+
+    @Override
+    public void dropColumn(ExternalTable dorisTable, String columnName, long updateTime) throws UserException {
+        alterTable(dorisTable, Collections.singletonList(SchemaChange.dropColumn(columnName)),
+                "drop column " + columnName);
+        refreshTable(dorisTable, updateTime);
+    }
+
+    @Override
+    public void renameColumn(ExternalTable dorisTable, String oldName, String newName, long updateTime)
+            throws UserException {
+        alterTable(dorisTable, Collections.singletonList(SchemaChange.renameColumn(oldName, newName)),
+                "rename column " + oldName + " to " + newName);
+        refreshTable(dorisTable, updateTime);
+    }
+
+    @Override
+    public void updateTableProperties(ExternalTable dorisTable, Map<String, String> properties)
+            throws UserException {
+        List<SchemaChange> changes = properties.entrySet().stream()
+                .map(entry -> SchemaChange.setOption(entry.getKey(), entry.getValue()))
+                .collect(Collectors.toList());
+        alterTable(dorisTable, changes, "update properties");
+    }
+
     public Catalog getCatalog() {
         return catalog;
     }

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/ModifyTablePropertiesOp.java

@@ -90,6 +90,11 @@ public void validate(ConnectContext ctx) throws UserException {
             throw new AnalysisException("Properties is not set");
         }
 
+        if (tableName != null && !InternalCatalog.INTERNAL_CATALOG_NAME.equals(tableName.getCtl())) {
+            validateExternalTableProperties();
+            return;
+        }
+
         if (properties.size() != 1
                 && !TableProperty.isSamePrefixProperties(
                         properties, DynamicPartitionProperty.DYNAMIC_PARTITION_PROPERTY_PREFIX)
@@ -375,6 +380,25 @@ public void validate(ConnectContext ctx) throws UserException {
         analyzeForMTMV();
     }
 
+    private void validateExternalTableProperties() throws AnalysisException {
+        if (!properties.containsKey(PropertyAnalyzer.PROPERTIES_AUTO_ANALYZE_POLICY)) {
+            return;
+        }
+        if (properties.size() != 1) {
+            throw new AnalysisException("auto_analyze_policy cannot be set with external table properties");
+        }
+        String analyzePolicy = properties.get(PropertyAnalyzer.PROPERTIES_AUTO_ANALYZE_POLICY);
+        if (!PropertyAnalyzer.ENABLE_AUTO_ANALYZE_POLICY.equals(analyzePolicy)
+                && !PropertyAnalyzer.DISABLE_AUTO_ANALYZE_POLICY.equals(analyzePolicy)
+                && !PropertyAnalyzer.USE_CATALOG_AUTO_ANALYZE_POLICY.equals(analyzePolicy)) {
+            throw new AnalysisException(
+                    "Table auto analyze policy only support for " + PropertyAnalyzer.ENABLE_AUTO_ANALYZE_POLICY
+                            + " or " + PropertyAnalyzer.DISABLE_AUTO_ANALYZE_POLICY
+                            + " or " + PropertyAnalyzer.USE_CATALOG_AUTO_ANALYZE_POLICY);
+        }
+        this.opType = AlterOpType.MODIFY_TABLE_PROPERTY_SYNC;
+    }
+
     private void analyzeForMTMV() throws AnalysisException {
         if (tableName != null) {
             // Skip external catalog.

fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonMetadataOpsTest.java

@@ -17,9 +17,13 @@
 
 package org.apache.doris.datasource.paimon;
 
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.Type;
+import org.apache.doris.catalog.info.ColumnPosition;
 import org.apache.doris.common.DdlException;
 import org.apache.doris.common.UserException;
 import org.apache.doris.datasource.CatalogFactory;
+import org.apache.doris.datasource.ExternalTable;
 import org.apache.doris.nereids.parser.NereidsParser;
 import org.apache.doris.nereids.trees.plans.commands.CreateCatalogCommand;
 import org.apache.doris.nereids.trees.plans.commands.CreateTableCommand;
@@ -50,6 +54,8 @@
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.UUID;
@@ -220,6 +226,36 @@ public void testBucket() throws Exception {
         Assert.assertEquals("c0", table.options().get("bucket-key"));
     }
 
+    @Test
+    public void testAlterTable() throws Exception {
+        String tableName = getTableName();
+        Identifier identifier = new Identifier(dbName, tableName);
+        createTable("create table " + dbName + "." + tableName + " (id int) engine = paimon");
+
+        PaimonExternalDatabase dorisDb = (PaimonExternalDatabase) paimonCatalog.getDbNullable(dbName);
+        ExternalTable dorisTable = new PaimonExternalTable(10000L, tableName, tableName, paimonCatalog, dorisDb);
+
+        ops.addColumn(dorisTable, new Column("name", Type.STRING, true), null, 1L);
+        ops.addColumn(dorisTable, new Column("age", Type.INT, true), ColumnPosition.FIRST, 2L);
+        ops.addColumns(dorisTable,
+                Arrays.asList(new Column("city", Type.STRING, true), new Column("score", Type.DOUBLE, true)), 3L);
+        Assert.assertEquals(Arrays.asList("age", "id", "name", "city", "score"),
+                ops.getCatalog().getTable(identifier).rowType().getFieldNames());
+
+        ops.renameColumn(dorisTable, "city", "location", 4L);
+        ops.dropColumn(dorisTable, "score", 5L);
+        Assert.assertEquals(Arrays.asList("age", "id", "name", "location"),
+                ops.getCatalog().getTable(identifier).rowType().getFieldNames());
+
+        ops.updateTableProperties(dorisTable, Collections.singletonMap("snapshot.num-retained.min", "3"));
+        Assert.assertEquals("3", ops.getCatalog().getTable(identifier).options().get("snapshot.num-retained.min"));
+
+        String newTableName = tableName + "_renamed";
+        ops.renameTableImpl(dbName, tableName, newTableName);
+        Assert.assertFalse(ops.tableExist(dbName, tableName));
+        Assert.assertTrue(ops.tableExist(dbName, newTableName));
+    }
+
     public void createTable(String sql) throws UserException {
         LogicalPlan plan = new NereidsParser().parseSingle(sql);
         Assertions.assertTrue(plan instanceof CreateTableCommand);

fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/info/ModifyTablePropertiesOpTest.java

@@ -0,0 +1,68 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.trees.plans.commands.info;
+
+import org.apache.doris.catalog.info.TableNameInfo;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.util.PropertyAnalyzer;
+import org.apache.doris.qe.ConnectContext;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class ModifyTablePropertiesOpTest {
+
+    @Test
+    public void testExternalTablePropertiesAllowConnectorOptions() throws Exception {
+        Map<String, String> properties = new HashMap<>();
+        properties.put("file.format", "parquet");
+        properties.put("write-buffer-size", "128 mb");
+        ModifyTablePropertiesOp op = new ModifyTablePropertiesOp(properties);
+        op.setTableName(new TableNameInfo("external", "db", "tbl"));
+
+        op.validate(new ConnectContext());
+    }
+
+    @Test
+    public void testExternalAutoAnalyzePolicyCannotBeMixedWithConnectorOptions() {
+        Map<String, String> properties = new HashMap<>();
+        properties.put(PropertyAnalyzer.PROPERTIES_AUTO_ANALYZE_POLICY,
+                PropertyAnalyzer.ENABLE_AUTO_ANALYZE_POLICY);
+        properties.put("file.format", "parquet");
+        ModifyTablePropertiesOp op = new ModifyTablePropertiesOp(properties);
+        op.setTableName(new TableNameInfo("external", "db", "tbl"));
+
+        AnalysisException exception = Assertions.assertThrows(
+                AnalysisException.class, () -> op.validate(new ConnectContext()));
+        Assertions.assertTrue(exception.getMessage().contains("cannot be set with external table properties"));
+    }
+
+    @Test
+    public void testExternalAutoAnalyzePolicyIsValidated() {
+        ModifyTablePropertiesOp op = new ModifyTablePropertiesOp(
+                java.util.Collections.singletonMap(PropertyAnalyzer.PROPERTIES_AUTO_ANALYZE_POLICY, "invalid"));
+        op.setTableName(new TableNameInfo("external", "db", "tbl"));
+
+        AnalysisException exception = Assertions.assertThrows(
+                AnalysisException.class, () -> op.validate(new ConnectContext()));
+        Assertions.assertTrue(exception.getMessage().contains("Table auto analyze policy only support"));
+    }
+}