private static TableLoader createTableLoader(
            CatalogBaseTable catalogBaseTable,
            Map<String, String> tableProps,
            String databaseName,
            String tableName) {
        Configuration flinkConf = new Configuration();
        tableProps.forEach(flinkConf::setString);

        String catalogName = flinkConf.getString(CATALOG_NAME);
        Preconditions.checkNotNull(
                catalogName, "Table property '%s' cannot be null", CATALOG_NAME.key());

        String catalogDatabase = flinkConf.getString(CATALOG_DATABASE, databaseName);
        Preconditions.checkNotNull(catalogDatabase, "The iceberg database name cannot be null");

        String catalogTable = flinkConf.getString(CATALOG_TABLE, tableName);
        Preconditions.checkNotNull(catalogTable, "The iceberg table name cannot be null");

        org.apache.hadoop.conf.Configuration hadoopConf = FlinkCatalogFactory.clusterHadoopConf();
        FlinkCatalogFactory factory = new FlinkCatalogFactory();
        FlinkCatalog flinkCatalog =
                (FlinkCatalog) factory.createCatalog(catalogName, tableProps, hadoopConf);
        ObjectPath objectPath = new ObjectPath(catalogDatabase, catalogTable);

        // Create database if not exists in the external catalog.
        if (!flinkCatalog.databaseExists(catalogDatabase)) {
            try {
                flinkCatalog.createDatabase(
                        catalogDatabase, new CatalogDatabaseImpl(Maps.newHashMap(), null), true);
            } catch (DatabaseAlreadyExistException e) {
                throw new AlreadyExistsException(
                        e,
                        "Database %s already exists in the iceberg catalog %s.",
                        catalogName,
                        catalogDatabase);
            }
        }

        // Create table if not exists in the external catalog.
        if (!flinkCatalog.tableExists(objectPath)) {
            try {
                flinkCatalog.createIcebergTable(objectPath, catalogBaseTable, true);
            } catch (TableAlreadyExistException e) {
                throw new AlreadyExistsException(
                        e,
                        "Table %s already exists in the database %s and catalog %s",
                        catalogTable,
                        catalogDatabase,
                        catalogName);
            }
        }

        return TableLoader.fromCatalog(
                flinkCatalog.getCatalogLoader(), TableIdentifier.of(catalogDatabase, catalogTable));
    }
Copy from iceberg-flink:iceberg-flink-1.15:1.3.1
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/FlinkDynamicTableFactory.java