PaimonMetadataOps.java

// 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.datasource.paimon;

import org.apache.doris.analysis.ColumnPosition;
import org.apache.doris.analysis.PartitionDesc;
import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.Type;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.ErrorCode;
import org.apache.doris.common.ErrorReport;
import org.apache.doris.common.UserException;
import org.apache.doris.common.security.authentication.ExecutionAuthenticator;
import org.apache.doris.common.util.Util;
import org.apache.doris.datasource.DorisTypeVisitor;
import org.apache.doris.datasource.ExternalCatalog;
import org.apache.doris.datasource.ExternalDatabase;
import org.apache.doris.datasource.ExternalTable;
import org.apache.doris.datasource.operations.ExternalMetadataOps;
import org.apache.doris.nereids.trees.plans.commands.info.ColumnDefinition;
import org.apache.doris.nereids.trees.plans.commands.info.CreateOrReplaceBranchInfo;
import org.apache.doris.nereids.trees.plans.commands.info.CreateOrReplaceTagInfo;
import org.apache.doris.nereids.trees.plans.commands.info.CreateTableInfo;
import org.apache.doris.nereids.trees.plans.commands.info.DropBranchInfo;
import org.apache.doris.nereids.trees.plans.commands.info.DropTagInfo;

import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.Catalog.DatabaseNotEmptyException;
import org.apache.paimon.catalog.Catalog.DatabaseNotExistException;
import org.apache.paimon.catalog.Catalog.TableAlreadyExistException;
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.DataField;
import org.apache.paimon.types.DataType;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;

public class PaimonMetadataOps implements ExternalMetadataOps {

    private static final Logger LOG = LogManager.getLogger(PaimonMetadataOps.class);
    protected Catalog catalog;
    protected ExternalCatalog dorisCatalog;
    private ExecutionAuthenticator executionAuthenticator;
    private static final String PRIMARY_KEY_IDENTIFIER = "primary-key";
    private static final String PROP_COMMENT = "comment";
    private static final String PROP_LOCATION = "location";

    public PaimonMetadataOps(ExternalCatalog dorisCatalog, Catalog catalog) {
        this.dorisCatalog = dorisCatalog;
        this.catalog = catalog;
        this.executionAuthenticator = dorisCatalog.getExecutionAuthenticator();
    }


    @Override
    public boolean createDbImpl(String dbName, boolean ifNotExists, Map<String, String> properties)
            throws DdlException {
        try {
            return executionAuthenticator.execute(() -> performCreateDb(dbName, ifNotExists, properties));
        } catch (Exception e) {
            throw new DdlException("Failed to create database: "
                + dbName + ": " + Util.getRootCauseMessage(e), e);
        }
    }

    private boolean performCreateDb(String dbName, boolean ifNotExists, Map<String, String> properties)
            throws DdlException, Catalog.DatabaseAlreadyExistException {
        if (databaseExist(dbName)) {
            if (ifNotExists) {
                LOG.info("create database[{}] which already exists", dbName);
                return true;
            } else {
                ErrorReport.reportDdlException(ErrorCode.ERR_DB_CREATE_EXISTS, dbName);
            }
        }

        if (!properties.isEmpty() && dorisCatalog instanceof PaimonExternalCatalog) {
            String catalogType = ((PaimonExternalCatalog) dorisCatalog).getCatalogType();
            if (!PaimonExternalCatalog.PAIMON_HMS.equals(catalogType)) {
                throw new DdlException(
                    "Not supported: create database with properties for paimon catalog type: " + catalogType);
            }
        }

        catalog.createDatabase(dbName, ifNotExists, properties);
        return false;
    }

    @Override
    public void afterCreateDb() {
        dorisCatalog.resetMetaCacheNames();
    }

    @Override
    public void dropDbImpl(String dbName, boolean ifExists, boolean force) throws DdlException {
        try {
            executionAuthenticator.execute(() -> {
                performDropDb(dbName, ifExists, force);
                return null;
            });
        } catch (Exception e) {
            throw new DdlException(
                "Failed to drop database: " + dbName + ", error message is:" + e.getMessage(), e);
        }
    }

    private void performDropDb(String dbName, boolean ifExists, boolean force) throws DdlException {
        ExternalDatabase dorisDb = dorisCatalog.getDbNullable(dbName);
        if (dorisDb == null) {
            if (ifExists) {
                LOG.info("drop database[{}] which does not exist", dbName);
                // Database does not exist and IF EXISTS is specified; treat as no-op.
                return;
            } else {
                ErrorReport.reportDdlException(ErrorCode.ERR_DB_DROP_EXISTS, dbName);
                // ErrorReport.reportDdlException is expected to throw DdlException.
                return;
            }
        }

        if (force) {
            List<String> tableNames = listTableNames(dbName);
            if (!tableNames.isEmpty()) {
                LOG.info("drop database[{}] with force, drop all tables, num: {}", dbName, tableNames.size());
            }
            for (String tableName : tableNames) {
                performDropTable(dbName, tableName, true);
            }
        }

        try {
            catalog.dropDatabase(dbName, ifExists, force);
        } catch (DatabaseNotExistException e) {
            throw new RuntimeException("database " + dbName + " does not exist!");
        } catch (DatabaseNotEmptyException e) {
            throw new RuntimeException("database " + dbName + " is not empty! please check!");
        }
    }

    @Override
    public void afterDropDb(String dbName) {
        dorisCatalog.unregisterDatabase(dbName);
    }

    @Override
    public boolean createTableImpl(CreateTableInfo createTableInfo) throws UserException {
        try {
            return executionAuthenticator.execute(() -> performCreateTable(createTableInfo));
        } catch (Exception e) {
            throw new DdlException(
                "Failed to create table: " + createTableInfo.getTableName() + ", error message is:" + e.getMessage(),
                e);
        }
    }

    public boolean performCreateTable(CreateTableInfo createTableInfo) throws UserException {
        String dbName = createTableInfo.getDbName();
        ExternalDatabase<?> db = dorisCatalog.getDbNullable(dbName);
        if (db == null) {
            throw new UserException("Failed to get database: '" + dbName + "' in catalog: " + dorisCatalog.getName());
        }
        String tableName = createTableInfo.getTableName();
        // 1. first, check if table exist in remote
        if (tableExist(db.getRemoteName(), tableName)) {
            if (createTableInfo.isIfNotExists()) {
                LOG.info("create table[{}] which already exists", tableName);
                return true;
            } else {
                ErrorReport.reportDdlException(ErrorCode.ERR_TABLE_EXISTS_ERROR, tableName);
            }
        }

        // 2. second, check if table exist in local.
        // This is because case sensibility issue, eg:
        // 1. lower_case_table_name = 1
        // 2. create table tbl1;
        // 3. create table TBL1;  TBL1 does not exist in remote because the remote system is case-sensitive.
        //    but because lower_case_table_name = 1, the table can not be created in Doris because it is conflict with
        //    tbl1
        ExternalTable dorisTable = db.getTableNullable(tableName);
        if (dorisTable != null) {
            if (createTableInfo.isIfNotExists()) {
                LOG.info("create table[{}] which already exists", tableName);
                return true;
            } else {
                ErrorReport.reportDdlException(ErrorCode.ERR_TABLE_EXISTS_ERROR, tableName);
            }
        }
        List<Column> columns = createTableInfo.getColumnDefinitions().stream()
                .map(ColumnDefinition::translateToCatalogStyle)
                .collect(Collectors.toList());
        Schema schema = toPaimonSchema(columns, createTableInfo.getPartitionDesc(),
                createTableInfo.getProperties());
        try {
            catalog.createTable(new Identifier(createTableInfo.getDbName(), createTableInfo.getTableName()),
                    schema, createTableInfo.isIfNotExists());
        } catch (TableAlreadyExistException | DatabaseNotExistException e) {
            throw new RuntimeException(e);
        }
        return false;
    }

    private Schema toPaimonSchema(List<Column> columns, PartitionDesc partitionDesc, Map<String, String> properties) {
        Map<String, String> normalizedProperties = new HashMap<>(properties);
        normalizedProperties.remove(PRIMARY_KEY_IDENTIFIER);
        normalizedProperties.remove(PROP_COMMENT);
        if (normalizedProperties.containsKey(PROP_LOCATION)) {
            String path = normalizedProperties.remove(PROP_LOCATION);
            normalizedProperties.put(CoreOptions.PATH.key(), path);
        }

        String pkAsString = properties.get(PRIMARY_KEY_IDENTIFIER);
        List<String> primaryKeys = pkAsString == null ? Collections.emptyList() : Arrays.stream(pkAsString.split(","))
                .map(String::trim)
                .collect(Collectors.toList());
        List<String> partitionKeys = partitionDesc == null ? new ArrayList<>() : partitionDesc.getPartitionColNames();
        List<String> rootFieldNames = columns.stream().map(Column::getName).collect(Collectors.toList());
        primaryKeys = getPaimonColumnNames(rootFieldNames, primaryKeys);
        partitionKeys = getPaimonColumnNames(rootFieldNames, partitionKeys);
        Schema.Builder schemaBuilder = Schema.newBuilder()
                .options(normalizedProperties)
                .primaryKey(primaryKeys)
                .partitionKeys(partitionKeys)
                .comment(properties.getOrDefault(PROP_COMMENT, null));
        for (Column column : columns) {
            schemaBuilder.column(column.getName(),
                    toPaimonType(column.getType()).copy(column.isAllowNull()),
                    column.getComment(),
                    column.getDefaultValue());
        }
        return schemaBuilder.build();
    }

    private List<String> getPaimonColumnNames(List<String> paimonColumnNames, List<String> dorisColumnNames) {
        Map<String, String> paimonColumnNameMap = paimonColumnNames.stream()
                .collect(Collectors.toMap(name -> name.toLowerCase(Locale.ROOT), name -> name));
        return dorisColumnNames.stream()
                .map(name -> paimonColumnNameMap.getOrDefault(name.toLowerCase(Locale.ROOT), name))
                .collect(Collectors.toList());
    }

    private DataType toPaimonType(Type type) {
        return DorisTypeVisitor.visit(type, new DorisToPaimonTypeVisitor());
    }

    @Override
    public void afterCreateTable(String dbName, String tblName) {
        Optional<ExternalDatabase<?>> db = dorisCatalog.getDbForReplay(dbName);
        if (db.isPresent()) {
            db.get().resetMetaCacheNames();
        }
        LOG.info("after create table {}.{}.{}, is db exists: {}",
                dorisCatalog.getName(), dbName, tblName, db.isPresent());
    }

    @Override
    public void dropTableImpl(ExternalTable dorisTable, boolean ifExists) throws DdlException {
        try {
            executionAuthenticator.execute(() -> {
                performDropTable(dorisTable.getRemoteDbName(), dorisTable.getRemoteName(), ifExists);
                return null;
            });
        } catch (Exception e) {
            throw new DdlException(
                "Failed to drop table: " + dorisTable.getName() + ", error message is:" + e.getMessage(), e);
        }
    }

    private void performDropTable(String dBName, String tableName, boolean ifExists) throws DdlException {
        if (!tableExist(dBName, tableName)) {
            if (ifExists) {
                LOG.info("drop table[{}] which does not exist", tableName);
                return;
            } else {
                ErrorReport.reportDdlException(ErrorCode.ERR_UNKNOWN_TABLE, tableName, dBName);
            }
        }
        try {
            catalog.dropTable(Identifier.create(dBName, tableName), ifExists);
        } catch (TableNotExistException e) {
            throw new RuntimeException("table " + tableName + " does not exist");
        }
    }

    @Override
    public void afterDropTable(String dbName, String tblName) {
        Optional<ExternalDatabase<?>> db = dorisCatalog.getDbForReplay(dbName);
        db.ifPresent(externalDatabase -> externalDatabase.unregisterTable(tblName));
        LOG.info("after drop table {}.{}.{}. is db exists: {}",
                dorisCatalog.getName(), dbName, tblName, db.isPresent());
    }

    @Override
    public void truncateTableImpl(ExternalTable dorisTable, List<String> partitions) throws DdlException {
        throw new UnsupportedOperationException("truncate table is not a supported operation!");
    }

    @Override
    public void createOrReplaceBranchImpl(ExternalTable dorisTable, CreateOrReplaceBranchInfo branchInfo)
            throws UserException {
        throw new UnsupportedOperationException("create or replace branch is not a supported operation!");
    }

    @Override
    public void createOrReplaceTagImpl(ExternalTable dorisTable, CreateOrReplaceTagInfo tagInfo) throws UserException {
        throw new UnsupportedOperationException("create or replace tag is not a supported operation!");
    }

    @Override
    public void dropTagImpl(ExternalTable dorisTable, DropTagInfo tagInfo) throws UserException {
        throw new UnsupportedOperationException("drop tag is not a supported operation!");
    }

    @Override
    public void dropBranchImpl(ExternalTable dorisTable, DropBranchInfo branchInfo) throws UserException {
        throw new UnsupportedOperationException("drop branch is not a supported operation!");
    }

    @Override
    public List<String> listDatabaseNames() {
        try {
            return executionAuthenticator.execute(() -> new ArrayList<>(catalog.listDatabases()));
        } catch (Exception e) {
            throw new RuntimeException("Failed to list databases names, catalog name: " + dorisCatalog.getName(), e);
        }
    }

    @Override
    public List<String> listTableNames(String db) {
        try {
            return executionAuthenticator.execute(() -> {
                List<String> tableNames = new ArrayList<>();
                try {
                    tableNames.addAll(catalog.listTables(db));
                } catch (DatabaseNotExistException e) {
                    LOG.warn("DatabaseNotExistException", e);
                }
                return tableNames;
            });
        } catch (Exception e) {
            throw new RuntimeException("Failed to list table names, catalog name: " + dorisCatalog.getName(), e);
        }
    }

    @Override
    public boolean tableExist(String dbName, String tblName) {
        try {
            return executionAuthenticator.execute(() -> {
                try {
                    catalog.getTable(Identifier.create(dbName, tblName));
                    return true;
                } catch (TableNotExistException e) {
                    return false;
                }
            });

        } catch (Exception e) {
            throw new RuntimeException("Failed to check table existence, catalog name: " + dorisCatalog.getName()
                + "error message is:" + ExceptionUtils.getRootCauseMessage(e), e);
        }
    }

    @Override
    public boolean databaseExist(String dbName) {
        try {
            return executionAuthenticator.execute(() -> {
                try {
                    catalog.getDatabase(dbName);
                    return true;
                } catch (DatabaseNotExistException e) {
                    return false;
                }
            });
        } catch (Exception e) {
            throw new RuntimeException("Failed to check database exist, error message is:" + e.getMessage(), e);
        }
    }

    private Identifier tableIdentifier(ExternalTable dorisTable) {
        return Identifier.create(dorisTable.getRemoteDbName(), dorisTable.getRemoteName());
    }

    private List<DataField> loadRemoteFields(ExternalTable dorisTable) throws UserException {
        try {
            return executionAuthenticator.execute(
                    () -> new ArrayList<>(catalog.getTable(tableIdentifier(dorisTable)).rowType().getFields()));
        } catch (Exception e) {
            throw new UserException("Failed to load schema for Paimon table " + dorisTable.getName()
                    + ": " + ExceptionUtils.getRootCauseMessage(e), e);
        }
    }

    private Map<String, DataField> indexFieldsByDorisName(List<DataField> fields) throws UserException {
        Map<String, DataField> fieldsByLowerCase = new HashMap<>();
        for (DataField field : fields) {
            DataField previous = fieldsByLowerCase.put(field.name().toLowerCase(Locale.ROOT), field);
            if (previous != null) {
                throw new UserException("Paimon table contains columns which differ only by case: "
                        + previous.name() + " and " + field.name());
            }
        }
        return fieldsByLowerCase;
    }

    private DataField resolveRemoteField(Map<String, DataField> fieldsByDorisName, String columnName)
            throws UserException {
        DataField field = fieldsByDorisName.get(columnName.toLowerCase(Locale.ROOT));
        if (field == null) {
            throw new UserException("Column " + columnName + " does not exist in Paimon table");
        }
        return field;
    }

    private DataType toPaimonColumnType(Column column) throws UserException {
        try {
            return toPaimonType(column.getType()).copy(column.isAllowNull());
        } catch (RuntimeException e) {
            throw new UserException("Unsupported Paimon type for column " + column.getName()
                    + ": " + ExceptionUtils.getRootCauseMessage(e), e);
        }
    }

    private void registerDorisColumnName(Set<String> columnNames, String columnName) throws UserException {
        if (!columnNames.add(columnName.toLowerCase(Locale.ROOT))) {
            throw new UserException("Column " + columnName
                    + " conflicts with an existing Paimon column (case-insensitive)");
        }
    }

    private void checkUnsupportedColumnAttributes(Column column) throws UserException {
        if (column.isAggregated()) {
            throw new UserException("Paimon column does not support aggregation method: " + column.getName());
        }
        if (column.isAutoInc()) {
            throw new UserException("Paimon column does not support AUTO_INCREMENT: " + column.getName());
        }
        if (column.isGeneratedColumn()) {
            throw new UserException("Column " + column.getName()
                    + " cannot be a generated column in a Paimon table");
        }
    }

    private void appendAddColumnChanges(List<SchemaChange> changes, Column column, SchemaChange.Move move)
            throws UserException {
        changes.add(SchemaChange.addColumn(
                column.getName(), toPaimonColumnType(column), column.getComment(), move));
        if (column.getDefaultValue() != null) {
            changes.add(SchemaChange.updateColumnDefaultValue(
                    new String[] {column.getName()}, column.getDefaultValue()));
        }
    }

    private void alterTable(ExternalTable dorisTable, List<SchemaChange> changes, String operation,
            long updateTime) throws UserException {
        if (changes.isEmpty()) {
            return;
        }
        try {
            executionAuthenticator.execute(() -> {
                catalog.alterTable(tableIdentifier(dorisTable), changes, false);
                return null;
            });
        } catch (Exception e) {
            throw new UserException("Failed to " + operation + " for Paimon table " + dorisTable.getName()
                    + ": " + ExceptionUtils.getRootCauseMessage(e), e);
        }
        refreshTable(dorisTable, updateTime);
    }

    private void refreshTable(ExternalTable dorisTable, long updateTime) {
        Optional<ExternalDatabase<?>> db = dorisCatalog.getDbForReplay(dorisTable.getDbName());
        if (db.isPresent()) {
            Optional<?> table = db.get().getTableForReplay(dorisTable.getName());
            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 {
        List<DataField> fields = loadRemoteFields(dorisTable);
        Map<String, DataField> fieldsByDorisName = indexFieldsByDorisName(fields);
        registerDorisColumnName(new HashSet<>(fieldsByDorisName.keySet()), column.getName());
        checkUnsupportedColumnAttributes(column);

        SchemaChange.Move move = null;
        if (position != null) {
            move = position.isFirst()
                    ? SchemaChange.Move.first(column.getName())
                    : SchemaChange.Move.after(column.getName(),
                            resolveRemoteField(fieldsByDorisName, position.getLastCol()).name());
        }

        List<SchemaChange> changes = new ArrayList<>();
        appendAddColumnChanges(changes, column, move);
        alterTable(dorisTable, changes, "add column " + column.getName(), updateTime);
    }

    @Override
    public void addColumns(ExternalTable dorisTable, List<Column> columns, long updateTime) throws UserException {
        Map<String, DataField> fieldsByDorisName = indexFieldsByDorisName(loadRemoteFields(dorisTable));
        Set<String> columnNames = new HashSet<>(fieldsByDorisName.keySet());
        List<SchemaChange> changes = new ArrayList<>();
        for (Column column : columns) {
            registerDorisColumnName(columnNames, column.getName());
            checkUnsupportedColumnAttributes(column);
            appendAddColumnChanges(changes, column, null);
        }
        alterTable(dorisTable, changes, "add columns", updateTime);
    }

    @Override
    public void dropColumn(ExternalTable dorisTable, String columnName, long updateTime) throws UserException {
        Map<String, DataField> fieldsByDorisName = indexFieldsByDorisName(loadRemoteFields(dorisTable));
        String remoteColumnName = resolveRemoteField(fieldsByDorisName, columnName).name();
        alterTable(dorisTable, Collections.singletonList(SchemaChange.dropColumn(remoteColumnName)),
                "drop column " + remoteColumnName, updateTime);
    }

    @Override
    public void renameColumn(ExternalTable dorisTable, String oldName, String newName, long updateTime)
            throws UserException {
        Map<String, DataField> fieldsByDorisName = indexFieldsByDorisName(loadRemoteFields(dorisTable));
        DataField oldField = resolveRemoteField(fieldsByDorisName, oldName);
        DataField conflictingField = fieldsByDorisName.get(newName.toLowerCase(Locale.ROOT));
        if (conflictingField != null && conflictingField != oldField) {
            throw new UserException("Column " + newName
                    + " conflicts with an existing Paimon column (case-insensitive)");
        }
        if (oldField.name().equals(newName)) {
            return;
        }
        alterTable(dorisTable,
                Collections.singletonList(SchemaChange.renameColumn(oldField.name(), newName)),
                "rename column " + oldField.name() + " to " + newName, updateTime);
    }

    @Override
    public void modifyColumn(ExternalTable dorisTable, Column column, ColumnPosition position, long updateTime)
            throws UserException {
        checkUnsupportedColumnAttributes(column);
        Map<String, DataField> fieldsByDorisName = indexFieldsByDorisName(loadRemoteFields(dorisTable));
        DataField currentField = resolveRemoteField(fieldsByDorisName, column.getName());
        DataType requestedType = toPaimonColumnType(column);
        List<SchemaChange> changes = new ArrayList<>();

        DataType requestedTypeWithCurrentNullability =
                requestedType.copy(currentField.type().isNullable());
        if (!currentField.type().equalsIgnoreFieldId(requestedTypeWithCurrentNullability)) {
            changes.add(SchemaChange.updateColumnType(
                    currentField.name(), requestedTypeWithCurrentNullability, true));
        }
        if (currentField.type().isNullable() != requestedType.isNullable()) {
            changes.add(SchemaChange.updateColumnNullability(
                    currentField.name(), requestedType.isNullable()));
        }
        if (!Objects.equals(currentField.description(), column.getComment())) {
            changes.add(SchemaChange.updateColumnComment(currentField.name(), column.getComment()));
        }
        if (!Objects.equals(currentField.defaultValue(), column.getDefaultValue())) {
            changes.add(SchemaChange.updateColumnDefaultValue(
                    new String[] {currentField.name()}, column.getDefaultValue()));
        }
        if (position != null) {
            SchemaChange.Move move = position.isFirst()
                    ? SchemaChange.Move.first(currentField.name())
                    : SchemaChange.Move.after(currentField.name(),
                            resolveRemoteField(fieldsByDorisName, position.getLastCol()).name());
            changes.add(SchemaChange.updateColumnPosition(move));
        }

        alterTable(dorisTable, changes, "modify column " + currentField.name(), updateTime);
    }

    @Override
    public void reorderColumns(ExternalTable dorisTable, List<String> newOrder, long updateTime)
            throws UserException {
        List<DataField> fields = loadRemoteFields(dorisTable);
        Map<String, DataField> fieldsByDorisName = indexFieldsByDorisName(fields);
        if (newOrder.size() != fields.size()) {
            throw new UserException("Reorder columns must contain every Paimon column exactly once");
        }

        List<String> remoteOrder = new ArrayList<>(newOrder.size());
        Set<String> seen = new HashSet<>();
        for (String columnName : newOrder) {
            DataField field = resolveRemoteField(fieldsByDorisName, columnName);
            if (!seen.add(field.name().toLowerCase(Locale.ROOT))) {
                throw new UserException("Duplicate column in reorder columns: " + columnName);
            }
            remoteOrder.add(field.name());
        }
        List<String> currentOrder = fields.stream().map(DataField::name).collect(Collectors.toList());
        if (currentOrder.equals(remoteOrder)) {
            return;
        }

        List<SchemaChange> changes = new ArrayList<>();
        changes.add(SchemaChange.updateColumnPosition(SchemaChange.Move.first(remoteOrder.get(0))));
        for (int i = 1; i < remoteOrder.size(); i++) {
            changes.add(SchemaChange.updateColumnPosition(
                    SchemaChange.Move.after(remoteOrder.get(i), remoteOrder.get(i - 1))));
        }
        alterTable(dorisTable, changes, "reorder columns", updateTime);
    }

    public Catalog getCatalog() {
        return catalog;
    }

    @Override
    public void close() {
        if (catalog != null) {
            catalog = null;
        }
    }
}