class TDengineMetadataParser implements TableMetadataParser {

    private final RDBSchemaMetadata schema;

    private ReactiveSqlExecutor sql() {
        return schema.findFeatureNow(ReactiveSqlExecutor.ID);
    }

    @Override
    public List<String> parseAllTableName() {
        throw new UnsupportedOperationException();
    }

    @Override
    public Flux<String> parseAllTableNameReactive() {
        return sql()
            .select("show stables", ResultWrappers.map())
            .mapNotNull(map -> (String) map.get("stable_name"));
    }

    @Override
    public boolean tableExists(String name) {
        throw new UnsupportedOperationException();
    }

    @Override
    public Mono<Boolean> tableExistsReactive(String name) {
        return parseAllTableNameReactive()
            .hasElement(name);
    }

    @Override
    public Optional<? extends ObjectMetadata> parseByName(String s) {
        throw new UnsupportedOperationException();
    }

    @Override
    public List<? extends ObjectMetadata> parseAll() {
        throw new UnsupportedOperationException();
    }

    @Override
    public Mono<RDBTableMetadata> parseByNameReactive(String name) {<FILL_FUNCTION_BODY>}

    private RDBColumnMetadata convertToColumn(Map<String, Object> columnInfo) {
        String note = (String) columnInfo.getOrDefault("Note", "");
        String column = (String) columnInfo.get("Field");
        String type = (String) columnInfo.get("Type");
        int length = CastUtils.castNumber(columnInfo.get("Length")).intValue();

        RDBColumnMetadata metadata = new RDBColumnMetadata();
        metadata.setName(column);
        metadata.setProperty(TDengineConstants.COLUMN_IS_TAG, "tag".equalsIgnoreCase(note));
        metadata.setLength(length);
        metadata.setType(schema.getDialect().convertDataType(type));
        return metadata;
    }

    @Override
    public Flux<RDBTableMetadata> parseAllReactive() {
        return parseAllTableNameReactive()
            .flatMap(this::parseByNameReactive);
    }
}

        RDBTableMetadata table = schema.newTable(name);
        return sql()
            .select("describe "+table.getFullName(), ResultWrappers.map())
            .doOnNext(column -> table.addColumn(convertToColumn(column)))
            .then(Mono.fromSupplier(() -> table.getColumns().isEmpty() ? null : table));
    
<no_super_class>