@Override
    public DynamicTableSink createDynamicTableSink(Context context) {
        ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
        CatalogTable catalogTable = context.getCatalogTable();
        Map<String, String> writeProps = catalogTable.getOptions();
        TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(catalogTable.getSchema());

        TableLoader tableLoader;
        if (catalog != null) {
            tableLoader = createTableLoader(catalog, objectPath);
        } else {
            tableLoader =
                    createTableLoader(
                            catalogTable, writeProps, objectPath.getDatabaseName(), objectPath.getObjectName());
        }
        return new IcebergTableSink(tableLoader, tableSchema, context.getConfiguration(), writeProps);
    }
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