@Override
    public DynamicTableSource createDynamicTableSource(Context context) {

        ObjectIdentifier objectIdentifier = context.getObjectIdentifier();
        CatalogTable catalogTable = context.getCatalogTable();
        Map<String, String> tableProps = catalogTable.getOptions();
        TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(catalogTable.getSchema());

        TableLoader tableLoader =
                createTableLoader(
                        catalogTable,
                        tableProps,
                        objectIdentifier.getDatabaseName(),
                        objectIdentifier.getObjectName());
        return new IcebergTableSource(tableLoader, tableSchema, tableProps, context.getConfiguration());
    }
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