FlinkSQL-基于jdbc的自定义Catalog

本文最后更新于：2023年8月15日 晚上

FlinkSQL-基于jdbc的自定义Catalog

本文将介绍如何在FlinkSQL中使用基于jdbc的自定义Catalog。

什么是Catalog？

Catalog是一个用于管理数据库和表的元数据存储系统。在FlinkSQL中，Catalog用于描述Flink集群中的数据库和表。FlinkSQL支持多种类型的Catalog，如默认基于内存的GenericInMemoryCatalog,基于Hive的HiveCatalog、基于JDBC的MySqlCatalog和PostgresCatalog等。

  1. GenericInMemoryCatalog:基于内存,所有数据库和表信息都存储在内存中,重启后即消失
  2. HiveCatalog:数据存储在hive的元数据中,需要部署hive服务才能用
  3. MySqlCatalog和PostgresCatalog: 基于jdbc,只能读取表结构,对应的jdbc数据库中有哪些表就只能用哪些表,无法创建。

自定义Catalog

在FlinkSQL中，除了使用默认的Catalog外，我们还可以通过实现自定义的Catalog来管理数据源和表。自定义Catalog可以更灵活地管理数据库和表，自定义持久化逻辑，满足不同场景下的需求。本文将介绍如何使用基于jdbc的自定义Catalog。

实现

首先，我们需要实现一个继承自JdbcCatalog的自定义Catalog。JdbcCatalog是Flink SQL中内置的一个基于JDBC的Catalog实现，我们可以通过继承该类，重写其中的方法来实现自定义的Catalog。

自定义Catalog需要继承JdbcCatalog类，并重写其中的方法来实现自定义的逻辑。在打开和关闭Catalog时，我们可以实现自定义的初始化和资源释放逻辑。在listTables和getTable方法中，我们需要实现自定义的表列表查询和表查询逻辑。除此之外，还可以重写其他方法来满足不同的需求。

设计表结构

  1. 基于mysql,如果要使用其他数据库相应调整ddl即可

  • 数据库(Database):

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    CREATE TABLE `metadata_database`
    (
        `id`            bigint(20)  NOT NULL AUTO_INCREMENT COMMENT '主键id',
        `create_person` varchar(100)         DEFAULT '' COMMENT '创建人',
        `create_time`   datetime             DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
        `update_person` varchar(100)         DEFAULT '' COMMENT '最后更新人',
        `update_time`   datetime             DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
        `dr`            tinyint(2)           DEFAULT '0' COMMENT '逻辑删除标记 0 未删除 1 已删除',
        `database_name` varchar(100)         DEFAULT NULL COMMENT '数据库名',
        `comment`       varchar(200)         DEFAULT NULL COMMENT '备注信息',
        PRIMARY KEY (`id`) USING BTREE,
        UNIQUE KEY `uni_database_name` (`database_name`, `type`)
    ) ENGINE = InnoDB
      DEFAULT CHARSET = utf8mb4
      ROW_FORMAT = DYNAMIC COMMENT ='元数据_数据库

  • 数据表(Table):

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    CREATE TABLE `metadata_table`
    (
        `id`                  bigint(20)   NOT NULL AUTO_INCREMENT COMMENT '主键id',
        `create_person`       varchar(100)          DEFAULT '' COMMENT '创建人',
        `create_time`         datetime              DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
        `update_person`       varchar(100)          DEFAULT '' COMMENT '最后更新人',
        `update_time`         datetime              DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
        `dr`                  tinyint(2)            DEFAULT '0' COMMENT '逻辑删除标记 0 未删除 1 已删除',
        `database_name`       varchar(100)          DEFAULT NULL COMMENT '数据库名',
        `database_id`         bigint(20)            DEFAULT NULL COMMENT '数据库id',
        `table_name`          varchar(100) NOT NULL,
        `comment`             varchar(200)          DEFAULT NULL COMMENT '备注',
        PRIMARY KEY (`id`) USING BTREE,
        UNIQUE KEY `idx_table_name` (`table_name`, `database_name`, `type`),
        KEY `idx_database_id` (`database_id`)
    ) ENGINE = InnoDB
      DEFAULT CHARSET = utf8mb4
      ROW_FORMAT = DYNAMIC COMMENT ='元数据_数据表';

  • 字段信息(Column)

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    CREATE TABLE `metadata_column`
    (
        `id`                  bigint(20) NOT NULL AUTO_INCREMENT COMMENT '主键id',
        `create_person`       varchar(100)        DEFAULT '' COMMENT '创建人',
        `create_time`         datetime            DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
        `update_person`       varchar(100)        DEFAULT '' COMMENT '最后更新人',
        `update_time`         datetime            DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
        `dr`                  tinyint(2)          DEFAULT '0' COMMENT '逻辑删除标记 0 未删除 1 已删除',
        `column_name`         varchar(100)        DEFAULT NULL COMMENT '字段名',
        `column_type`         varchar(100)        DEFAULT NULL COMMENT '字段类型',
        `column_length`       int(11)             DEFAULT NULL COMMENT '字段长度',
        `precision`           int(11)             DEFAULT NULL COMMENT '字段精度',
        `table_id`            bigint(20)          DEFAULT NULL COMMENT '表id',
        `database_id`         bigint(20)          DEFAULT NULL COMMENT '数据库id',
        `nullable`            tinyint(1)          DEFAULT NULL COMMENT '是否可空',
        `scale`               int(11)             DEFAULT NULL COMMENT '小数位数',
        `comment`             varchar(200)        DEFAULT NULL COMMENT '备注信息',
        `primary_key`         tinyint(1) NOT NULL DEFAULT '0' COMMENT '是否是主键',
        PRIMARY KEY (`id`) USING BTREE,
        KEY `idx_database_id` (`database_id`),
        KEY `idx_table_id` (`table_id`)
    ) ENGINE = InnoDB
      DEFAULT CHARSET = utf8mb4
      ROW_FORMAT = DYNAMIC COMMENT ='元数据_字段';

  • 属性信息(Properties)

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    CREATE TABLE `metadata_properties`
    (
        `id`            bigint(20) NOT NULL AUTO_INCREMENT COMMENT '主键id',
        `create_person` varchar(100) DEFAULT '' COMMENT '创建人',
        `create_time`   datetime     DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
        `update_person` varchar(100) DEFAULT '' COMMENT '最后更新人',
        `update_time`   datetime     DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
        `dr`            tinyint(2)   DEFAULT '0' COMMENT '逻辑删除标记 0 未删除 1 已删除',
        `key`           varchar(100) DEFAULT NULL COMMENT 'key',
        `value`         varchar(300) DEFAULT NULL COMMENT 'value',
        `data_id`       bigint(20)   DEFAULT NULL COMMENT '表id',
        `data_type`     varchar(20)  DEFAULT NULL COMMENT '数据类型',
        PRIMARY KEY (`id`) USING BTREE,
        UNIQUE KEY `uni_data_key` (`data_id`, `key`, `data_type`)
    ) ENGINE = InnoDB
      DEFAULT CHARSET = utf8mb4
      ROW_FORMAT = DYNAMIC COMMENT ='元数据_数据表属性';

  • 水印信息(Watermark)

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    CREATE TABLE `metadata_watermark`
    (
        `id`            bigint(20) NOT NULL AUTO_INCREMENT COMMENT '主键id',
        `create_person` varchar(100) DEFAULT '' COMMENT '创建人',
        `create_time`   datetime     DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
        `update_person` varchar(100) DEFAULT '' COMMENT '最后更新人',
        `update_time`   datetime     DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
        `dr`            tinyint(2)   DEFAULT '0' COMMENT '逻辑删除标记 0 未删除 1 已删除',
        `column_name`   varchar(100) DEFAULT NULL COMMENT '字段名',
        `expression`    varchar(500) DEFAULT NULL COMMENT '表达式',
        `table_id`      bigint(20)   DEFAULT NULL COMMENT '表id',
        `database_id`   bigint(20) NOT NULL COMMENT '数据库id',
        PRIMARY KEY (`id`)
    ) ENGINE = InnoDB
      DEFAULT CHARSET = utf8mb4 COMMENT ='元数据_数据表_水印配置';

  • 函数信息(Function)

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    CREATE TABLE `metadata_function`
    (
        `id`                 bigint(20)   NOT NULL AUTO_INCREMENT COMMENT '主键id',
        `create_person`      varchar(100)          DEFAULT '' COMMENT '创建人',
        `create_time`        datetime              DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
        `update_person`      varchar(100)          DEFAULT '' COMMENT '最后更新人',
        `update_time`        datetime              DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
        `dr`                 tinyint(2)   NOT NULL DEFAULT '0' COMMENT '逻辑删除标记 0 未删除 1 已删除',
        `class_name`         varchar(300) NOT NULL COMMENT '函数类名',
        `function_name`      varchar(100) NOT NULL COMMENT '函数名',
        `description`        varchar(500)          DEFAULT NULL COMMENT '函数描述',
        `detail_description` varchar(500)          DEFAULT NULL COMMENT '函数详细描述',
        `is_generic`         tinyint(1)   NOT NULL DEFAULT '0' COMMENT '是否泛型',
        `function_language`  varchar(100) NOT NULL COMMENT '函数语言',
        `database_name`      varchar(100) NOT NULL COMMENT '数据库名',
        `database_id`        bigint(20)   NOT NULL COMMENT '数据库id',
        PRIMARY KEY (`id`),
        UNIQUE KEY `uni_idx_function_name` (`function_name`, `database_name`)
    ) ENGINE = InnoDB
      DEFAULT CHARSET = utf8mb4 COMMENT ='元数据_函数'

编写代码

持久层

目前持久层基于基于spring的mybatis-plus,这样有个问题,就是不支持在sqlclient等交互式场景使用,只能在TableEnvironment中手动注册,后续文章中(如果有的话)会改成支持手动配置mybatis-plus

Eo和Mapper

eo和mapper就是基本的字段信息,没什么特殊的,用代码生成器自动生成机器

Service

service加了一些必要的方法:

DatabaseService
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
@Service
public class DatabaseServiceImpl extends ServiceImpl<DatabaseMapper, DatabaseEo> implements DatabaseService {


    private final PropertiesServiceImpl propertiesServiceImpl;
    private final TableServiceImpl tableServiceImpl;
    private final FunctionServiceImpl functionServiceImpl;

    @Lazy
    public DatabaseServiceImpl(PropertiesServiceImpl propertiesServiceImpl,
                               TableServiceImpl tableServiceImpl, FunctionServiceImpl functionServiceImpl) {
        this.propertiesServiceImpl = propertiesServiceImpl;
        this.tableServiceImpl = tableServiceImpl;
        this.functionServiceImpl = functionServiceImpl;
    }

    public long count(@NonNull String databaseName) {
        return lambdaQuery().eq(DatabaseEo::getDatabaseName, databaseName)
                .count();
    }

    @Override
    public List<String> listDatabaseNames() {

        return lambdaQuery().list()
                .stream()
                .map(DatabaseEo::getDatabaseName)
                .collect(Collectors.toList());
    }

    @Override
    public List<DatabaseDto> listDatabase(@NotNull DatabaseDto query) {

        DatabaseEo example = BeanUtil.copyProperties(query, DatabaseEo.class);

        return lambdaQuery(example).list()
                .stream()
                .map(e -> BeanUtil.copyProperties(e, DatabaseDto.class))
                .collect(Collectors.toList());
    }

    @Override
    public DatabaseDto getDatabase(@NonNull String databaseName) {
        DatabaseEo databaseEo = lambdaQuery().eq(DatabaseEo::getDatabaseName, databaseName)
                .one();
        if (databaseEo != null) {

            DatabaseDto databaseDto = BeanUtil.copyProperties(databaseEo, DatabaseDto.class);
            Map<String, String> properties = propertiesServiceImpl.lambdaQuery()
                    .eq(PropertiesEo::getDataType, PropertiesDataTypeEnum.DATABASE)
                    .eq(PropertiesEo::getDataId, databaseEo.getId())
                    .list()
                    .stream()
                    .collect(Collectors.toMap(PropertiesEo::getKey, PropertiesEo::getValue));
            databaseDto.setProperties(properties);
            return databaseDto;
        }
        return null;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void createDatabase(DatabaseDto reqDto) {

        createDatabase(reqDto, false);
    }

    @Transactional(rollbackFor = Exception.class)
    public void createDatabase(DatabaseDto reqDto, boolean ignoreIfExist) {

        boolean exists = lambdaQuery().eq(DatabaseEo::getDatabaseName, reqDto.getDatabaseName())
                .exists();
        if (exists) {
            if (!ignoreIfExist) {
                throw ExceptionCodeEnum.DATABASE_EXISTS.toException();
            }
        } else {

            DatabaseEo databaseEo = new DatabaseEo();
            BeanUtil.copyProperties(reqDto, databaseEo);
            save(databaseEo);

            propertiesServiceImpl.saveProperties(reqDto.getProperties(), PropertiesDataTypeEnum.DATABASE.name(), databaseEo.getId(), true);
        }

    }


    @Override
    @Transactional(rollbackFor = Exception.class)
    public void deleteDatabase(@NonNull String databaseName, boolean cascade) {

        DatabaseEo databaseEo = lambdaQuery().eq(DatabaseEo::getDatabaseName, databaseName)
                .one();
        if (databaseEo == null) {
            throw ExceptionCodeEnum.DATABASE_NOT_EXISTS.toException();
        }

        removeById(databaseEo.getId());

        List<TableEo> tables = tableServiceImpl.lambdaQuery()
            	.eq(TableEo::getDatabaseName, databaseName)
                .list();
        if (CollUtil.isNotEmpty(tables)) {
            if (!cascade) {
                throw DATABASE_HAS_TABLE.toException();
            } else {
                List<Long> tableIds = tables.stream()
                        .map(BaseEo::getId)
                        .collect(Collectors.toList());
                tableServiceImpl.removeByIds(tableIds);
            }
        }

        functionServiceImpl.dropFunctionByDatabase(databaseName);

    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void updateDatabase(DatabaseDto reqDto) {

        DatabaseEo oldDatabaseEo = lambdaQuery().eq(DatabaseEo::getDatabaseName, reqDto.getDatabaseName())
                .one();
        if (oldDatabaseEo == null) {
            throw ExceptionCodeEnum.DATABASE_EXISTS.toException();
        } else {

            DatabaseEo databaseEo = new DatabaseEo();
            BeanUtil.copyProperties(reqDto, databaseEo);
            oldDatabaseEo.setId(oldDatabaseEo.getId());
            updateById(oldDatabaseEo);

            propertiesServiceImpl.saveProperties(reqDto.getProperties(), PropertiesDataTypeEnum.DATABASE.name(), oldDatabaseEo.getId(), false);
        }

    }

}
ColumnService
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
@Service
public class ColumnServiceImpl extends ServiceImpl<ColumnMapper, ColumnEo> {


    @Transactional(rollbackFor = Exception.class)
    public void saveTableColumns(List<ColumnDto> columnDtoList, @NonNull Long databaseId, @NonNull Long tableId, boolean create) {

        if (!create) {
            lambdaUpdate().eq(ColumnEo::getTableId, tableId)
                    .remove();
        }
        if (CollUtil.isNotEmpty(columnDtoList)) {

            List<ColumnEo> columns = new ArrayList<>(columnDtoList.size());
            for (ColumnDto columnDto : columnDtoList) {

                ColumnEo columnEo = BeanUtil.copyProperties(columnDto, ColumnEo.class);
                columnEo.setTableId(tableId);
                columnEo.setDatabaseId(databaseId);
                columns.add(columnEo);
            }
            saveBatch(columns);
        }
    }

}
PropertiesService
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
@Service
public class PropertiesServiceImpl extends ServiceImpl<PropertiesMapper, PropertiesEo> {


    public void saveProperties(Map<String, String> properties, @NonNull String dataType, @NonNull Long dataId, boolean create) {

        if (!create) {
            lambdaUpdate().eq(PropertiesEo::getDataType, dataType)
                    .eq(PropertiesEo::getDataId, dataId)
                    .remove();
        }

        if (CollUtil.isNotEmpty(properties)) {
            List<PropertiesEo> propertyList = properties.entrySet()
                    .stream()
                    .map(e -> {
                                PropertiesEo propertiesEo = new PropertiesEo();
                                propertiesEo.setKey(e.getKey());
                                propertiesEo.setValue(e.getValue());
                                propertiesEo.setDataId(dataId);
                                propertiesEo.setDataType(dataType);
                                return propertiesEo;}
                    ).collect(Collectors.toList());
            saveBatch(propertyList);
        }
    }


}
TableService
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
@Service
public class TableServiceImpl extends ServiceImpl<TableMapper, TableEo> implements TableService {


    private final PropertiesServiceImpl propertiesServiceImpl;
    private final ColumnServiceImpl columnServiceImpl;
    private final DatabaseServiceImpl databaseServiceImpl;
    private final WatermarkServiceImpl watermarkServiceImpl;
    private final DatasourceServiceImpl datasourceServiceImpl;

    @Lazy
    public TableServiceImpl(PropertiesServiceImpl propertiesServiceImpl,
                            ColumnServiceImpl columnServiceImpl,
                            DatabaseServiceImpl databaseServiceImpl,
                            WatermarkServiceImpl watermarkServiceImpl,
                            DatasourceServiceImpl datasourceServiceImpl) {
        this.propertiesServiceImpl = propertiesServiceImpl;
        this.columnServiceImpl = columnServiceImpl;
        this.databaseServiceImpl = databaseServiceImpl;
        this.watermarkServiceImpl = watermarkServiceImpl;
        this.datasourceServiceImpl = datasourceServiceImpl;
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public boolean removeByIds(Collection<?> list) {

        if (CollUtil.isNotEmpty(list)) {
            propertiesServiceImpl.lambdaUpdate()
                    .eq(PropertiesEo::getDataType, PropertiesDataTypeEnum.TABLE.name())
                    .in(PropertiesEo::getDataId, list)
                    .remove();
            columnServiceImpl.lambdaUpdate()
                  .in(ColumnEo::getTableId, list)
                    .remove();
            watermarkServiceImpl.lambdaUpdate()
                    .in(WatermarkEo::getTableId, list)
                    .remove();
            
            return super.removeByIds(list);
        }
        return true;

    }

    @Override
    public List<String> listTableNames(@NonNull String databaseName) {
        return lambdaQuery().eq(TableEo::getDatabaseName, databaseName)
                .list()
                .stream()
                .map(TableEo::getTableName)
                .collect(Collectors.toList());
    }

    @Override
    public List<TableDto> listTable(@NotNull TableDto query) {

        TableEo example = BeanUtil.copyProperties(query, TableEo.class);

        return lambdaQuery(example)
                .list()
                .stream()
                .map(e -> BeanUtil.copyProperties(e, TableDto.class))
                .collect(Collectors.toList());
    }

    @Override
    public TableDto getTable(@NonNull String databaseName, @NonNull String tableName) {

        TableEo tableEo = lambdaQuery()
            	.eq(TableEo::getDatabaseName, databaseName)
                .eq(TableEo::getTableName, tableName)
                .eq(TableEo::getDatasourceId, datasourceId)
                .one();
        if (tableEo != null) {

            return buildTableDto(tableEo);
        }

        return null;
    }

    private TableDto buildTableDto(TableEo tableEo) {
        TableDto tableDto = BeanUtil.copyProperties(tableEo, TableDto.class);
        Map<String, String> properties = propertiesServiceImpl.lambdaQuery()
                .eq(PropertiesEo::getDataType, PropertiesDataTypeEnum.TABLE)
                .eq(PropertiesEo::getDataId, tableEo.getId())
                .select(PropertiesEo::getKey, PropertiesEo::getValue)
                .list()
                .stream()
                .collect(Collectors.toMap(PropertiesEo::getKey, PropertiesEo::getValue));
        tableDto.setProperties(properties);

        List<ColumnDto> columnList = columnServiceImpl.lambdaQuery()
                .eq(ColumnEo::getTableId, tableEo.getId())
                .list()
                .stream()
                .map(e -> BeanUtil.copyProperties(e, ColumnDto.class))
                .collect(Collectors.toList());
        tableDto.setColumns(columnList);

        List<WatermarkDto> watermarks = watermarkServiceImpl.lambdaQuery()
                .eq(WatermarkEo::getTableId, tableEo.getId())
                .list()
                .stream()
                .map(e -> BeanUtil.copyProperties(e, WatermarkDto.class))
                .collect(Collectors.toList());
        tableDto.setWatermarks(watermarks);

        return tableDto;
    }

    @Override
    public boolean exists(@NonNull String databaseName, @NonNull String tableName) {
        return lambdaQuery()
            	.eq(TableEo::getDatabaseName, databaseName)
                .eq(TableEo::getTableName, tableName)
                .exists();
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void deleteTable(@NonNull String databaseName, @NonNull String tableName) {

        TableEo tableEo = lambdaQuery()
                .eq(TableEo::getDatabaseName, databaseName)
                .eq(TableEo::getTableName, tableName)
                .one();
        if (tableEo == null) {
            throw TABLE_NOT_EXISTS.toException();
        }

        removeById(tableEo.getId());

        propertiesServiceImpl.lambdaUpdate()
                .eq(PropertiesEo::getDataType, PropertiesDataTypeEnum.TABLE.name())
                .eq(PropertiesEo::getDataId, tableEo.getId())
                .remove();
        columnServiceImpl.lambdaUpdate()
                .eq(ColumnEo::getTableId, tableEo.getId())
                .remove();
        watermarkServiceImpl.lambdaUpdate()
                .eq(WatermarkEo::getTableId, tableEo.getId())
                .remove();
    }

    @Transactional(rollbackFor = Exception.class)
    public void renameTable(@NonNull String databaseName,
                            @NonNull String tableName,
                            @NonNull String newTableName) {

        TableEo tableEo = lambdaQuery()
                .eq(TableEo::getDatabaseName, databaseName)
                .eq(TableEo::getTableName, tableName)
                .one();
        if (tableEo == null) {
            throw TABLE_NOT_EXISTS.toException();
        }
        boolean newTableNameExists = lambdaQuery()
                .eq(TableEo::getDatabaseName, databaseName)
                .eq(TableEo::getTableName, newTableName)
                .ne(TableEo::getId, tableEo.getId())
                .exists();
        if (newTableNameExists) {
            throw ExceptionCodeEnum.TABLE_NAME_DUPLICATED.toException();
        }

        lambdaUpdate().set(TableEo::getTableName, newTableName)
                .eq(TableEo::getId, tableEo.getId())
                .update(new TableEo());
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Long createTable(@NonNull @Validated TableDto tableDto, boolean ignoreIfExist) {

        DatabaseEo databaseEo = databaseServiceImpl.lambdaQuery()
                .eq(DatabaseEo::getDatabaseName, tableDto.getDatabaseName())
                .one();
        if (databaseEo == null) {
            throw ExceptionCodeEnum.DATABASE_NOT_EXISTS.toException();
        }

        TableEo existTableEo = lambdaQuery()
                .eq(TableEo::getTableName, tableDto.getTableName())
                .eq(TableEo::getDatabaseName, tableDto.getDatabaseName())
                .one();
        if (existTableEo != null) {
            if(ignoreIfExist){
                return existTableEo.getId();
            }else {
                throw ExceptionCodeEnum.TABLE_NAME_DUPLICATED.toException();
            }
        }

        TableEo tableEo = BeanUtil.copyProperties(tableDto, TableEo.class);
        tableEo.setDatabaseId(databaseEo.getId());
        saveOrUpdate(tableEo);
        columnServiceImpl.saveTableColumns(tableDto.getColumns(), databaseEo.getId(), tableEo.getId(), true);
        watermarkServiceImpl.saveTableWatermarks(tableDto.getWatermarks(), databaseEo.getId(), tableEo.getId(), true);
        propertiesServiceImpl.saveProperties(tableDto.getProperties(), PropertiesDataTypeEnum.TABLE.name(), tableEo.getId(), true);

        return tableEo.getId();
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Long updateTable(@NonNull @Validated TableDto tableDto) {

        TableEo oldTableEo = lambdaQuery().eq(TableEo::getDatabaseName, tableDto.getDatabaseName())
                .eq(TableEo::getTableName, tableDto.getTableName())
                
                .one();
        if (oldTableEo == null) {
            throw TABLE_NOT_EXISTS.toException();
        }

        TableEo tableEo = BeanUtil.copyProperties(tableDto, TableEo.class);
        tableEo.setDatabaseId(oldTableEo.getDatabaseId());
        tableEo.setId(oldTableEo.getId());
        saveOrUpdate(tableEo);
        columnServiceImpl.saveTableColumns(tableDto.getColumns(), oldTableEo.getDatabaseId(), oldTableEo.getId(), false);
        watermarkServiceImpl.saveTableWatermarks(tableDto.getWatermarks(), oldTableEo.getDatabaseId(), oldTableEo.getId(), false);
        propertiesServiceImpl.saveProperties(tableDto.getProperties(), PropertiesDataTypeEnum.TABLE.name(), oldTableEo.getId(), false);

        return oldTableEo.getId();
    }

    @Override
    public TableDto getTable(@NotNull Long tableId) {
        TableEo tableEo = getById(tableId);
        if (tableEo != null) {
            return buildTableDto(tableEo);
        }
        return null;
    }
}

WatermarkService
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
@Service
public class WatermarkServiceImpl extends ServiceImpl<WatermarkMapper, WatermarkEo> {

    public void saveTableWatermarks(List<WatermarkDto> watermarkDtoList, @NonNull Long databaseId, @NonNull Long tableId, boolean create) {

        if (!create) {
            lambdaUpdate().eq(WatermarkEo::getTableId, tableId)
                    .remove();
        }

        if (CollUtil.isNotEmpty(watermarkDtoList)) {

            List<WatermarkEo> watermarks = new ArrayList<>(watermarkDtoList.size());
            for (WatermarkDto watermarkDto : watermarkDtoList) {
                WatermarkEo watermarkEo = BeanUtil.copyProperties(watermarkDto, WatermarkEo.class);
                watermarkEo.setTableId(tableId);
                watermarkEo.setDatabaseId(databaseId);
                watermarks.add(watermarkEo);
            }
            saveBatch(watermarks);
        }
    }

}
FunctionService
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
@Service
public class FunctionServiceImpl extends ServiceImpl<FunctionMapper, FunctionEo> {

    private final DatabaseServiceImpl databaseServiceImpl;

    public FunctionServiceImpl(DatabaseServiceImpl databaseServiceImpl) {
        this.databaseServiceImpl = databaseServiceImpl;
    }

    public List<String> listFunctionNames(@NotBlank String databaseName) {

        return lambdaQuery()
                .eq(FunctionEo::getDatabaseName, databaseName)
                .list()
                .stream()
                .map(FunctionEo::getFunctionName)
                .collect(Collectors.toList());
    }

    public FunctionDto getFunction(@NotBlank String databaseName, @NotBlank String functionName) {

        return lambdaQuery()
                .eq(FunctionEo::getDatabaseName, databaseName)
                .eq(FunctionEo::getFunctionName, functionName)
                .oneOpt()
                .map(this::buildFunctionDto)
                .orElse(null);
    }

    private FunctionDto buildFunctionDto(FunctionEo functionEo) {

        return BeanUtil.copyProperties(functionEo, FunctionDto.class);

    }

    public boolean exists(@NotBlank String databaseName, @NotBlank String objectName) {
        return lambdaQuery()
                .eq(FunctionEo::getDatabaseName, databaseName)
                .eq(FunctionEo::getFunctionName, objectName)
                .count() > 0;
    }

    public Long createFunction(@NotNull FunctionDto functionDto, boolean ignoreIfExists) {
        DatabaseEo databaseEo = databaseServiceImpl.lambdaQuery()
                .eq(DatabaseEo::getType, TableTypeEnum.FLINK.name())
                .eq(DatabaseEo::getDatabaseName, functionDto.getDatabaseName())
                .one();
        if (databaseEo == null) {
            throw ExceptionCodeEnum.DATABASE_NOT_EXISTS.toException();
        }

        FunctionEo existFunctionEo = lambdaQuery().eq(FunctionEo::getFunctionName, functionDto.getFunctionName())
                .eq(FunctionEo::getDatabaseId, databaseEo.getId())
                .one();
        if (existFunctionEo != null) {
            if (ignoreIfExists) {
                return existFunctionEo.getId();
            } else {
                throw ExceptionCodeEnum.TABLE_NAME_DUPLICATED.toException();
            }
        }

        FunctionEo functionEo = BeanUtil.copyProperties(functionDto, FunctionEo.class);
        functionEo.setDatabaseId(databaseEo.getId());
        save(functionEo);
        return functionEo.getId();
    }

    public void deleteFunction(@NotBlank String databaseName, @NotBlank String functionName) {

        FunctionEo functionEo = lambdaQuery()
                .eq(FunctionEo::getDatabaseName, databaseName)
                .eq(FunctionEo::getFunctionName, functionName)
                .one();
        if (functionEo == null) {
            throw ExceptionCodeEnum.FUNCTION_NOT_EXISTS.toException();
        }
        removeById(functionEo.getId());
    }

    public void updateFunction(@NotNull FunctionDto functionDto, boolean ignoreIfNotExists) {

        FunctionEo functionEo = lambdaQuery()
                .eq(FunctionEo::getDatabaseName, functionDto.getDatabaseName())
                .eq(FunctionEo::getFunctionName, functionDto.getFunctionName())
                .one();
        if (functionEo == null) {
            if (ignoreIfNotExists) {
                return;
            } else {
                throw ExceptionCodeEnum.FUNCTION_NOT_EXISTS.toException();
            }
        }
        //只允许修改函数类名和函数语言,描述信息
        functionEo.setClassName(functionDto.getClassName());
        functionEo.setFunctionLanguage(functionDto.getFunctionLanguage());
        functionEo.setDescription(functionDto.getDescription());
        functionEo.setDetailDescription(functionDto.getDetailDescription());
        updateById(functionEo);
    }

    public void dropFunctionByDatabase(@NotNull String databaseName) {

        lambdaUpdate().eq(FunctionEo::getDatabaseName, databaseName).remove();
    }
}
Catalog
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
package com.dtyunxi.dacat.metadata.catalog;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.lang.Assert;
import com.dtyunxi.dacat.core.config.FlinkProperties;
import com.dtyunxi.dacat.core.constant.TableTypeEnum;
import com.dtyunxi.dacat.core.dto.metadata.*;
import com.dtyunxi.dacat.core.exception.BusinessException;
import com.dtyunxi.dacat.metadata.service.DatabaseServiceImpl;
import com.dtyunxi.dacat.metadata.service.FunctionServiceImpl;
import com.dtyunxi.dacat.metadata.service.TableServiceImpl;
import com.dtyunxi.dacat.metadata.types.DataTypeMapper;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.catalog.*;
import org.apache.flink.table.catalog.exceptions.*;
import org.apache.flink.table.catalog.stats.CatalogColumnStatistics;
import org.apache.flink.table.catalog.stats.CatalogTableStatistics;
import org.apache.flink.table.expressions.Expression;
import org.apache.flink.table.expressions.SqlCallExpression;
import org.apache.flink.table.types.AbstractDataType;
import org.apache.flink.table.types.DataType;
import org.springframework.stereotype.Component;

import java.util.*;

import static com.dtyunxi.dacat.core.constant.CommonConstant.DATASOURCE_ID_FLINK;
import static com.dtyunxi.dacat.core.exception.ExceptionCodeEnum.*;

@Component
public class WritableJdbcCatalog extends AbstractCatalog {

    private final DatabaseServiceImpl databaseServiceImpl;
    private final TableServiceImpl tableServiceImpl;
    private final FunctionServiceImpl functionServiceImpl;

    public WritableJdbcCatalog(FlinkProperties flinkProperties,
                               DatabaseServiceImpl databaseServiceImpl,
                               TableServiceImpl tableServiceImpl,
                               FunctionServiceImpl functionServiceImpl) {
        super(flinkProperties.getCatalogName(), flinkProperties.getDefaultDatabase());
        this.databaseServiceImpl = databaseServiceImpl;
        this.tableServiceImpl = tableServiceImpl;
        this.createDefaultDatabase(flinkProperties.getDefaultDatabase());
        this.functionServiceImpl = functionServiceImpl;
    }

    private void createDefaultDatabase(String databaseName) {
        DatabaseDto databaseDto = new DatabaseDto();
        databaseDto.setDatabaseName(databaseName);
        databaseServiceImpl.createDatabase(databaseDto, true);
    }

    @Override
    public void open() throws CatalogException {
        //以mysql作为元数据的存储库,不需要open
    }

    @Override
    public void close() throws CatalogException {
        //以mysql作为元数据的存储库,不需要close
    }

    @Override
    public List<String> listDatabases() throws CatalogException {

        return databaseServiceImpl.listDatabaseNames();
    }

    @Override
    public CatalogDatabase getDatabase(String databaseName) throws DatabaseNotExistException, CatalogException {

        Assert.notBlank(databaseName, DATABASE_NAME_REQUIRED.getMsg());

        DatabaseDto database = databaseServiceImpl.getDatabase(databaseName);
        if (database != null) {
            return new CatalogDatabaseImpl(database.getProperties(), database.getComment());
        } else {
            throw new DatabaseNotExistException(getName(), databaseName);
        }
    }

    @Override
    public boolean databaseExists(String databaseName) throws CatalogException {
        Assert.notBlank(databaseName, DATABASE_NAME_REQUIRED.getMsg());
        return databaseServiceImpl.count(databaseName) > 0;
    }

    @Override
    public void createDatabase(String databaseName, CatalogDatabase database, boolean ignoreIfExists) throws DatabaseAlreadyExistException, CatalogException {

        Assert.notBlank(databaseName, DATABASE_NAME_REQUIRED.getMsg());
        Assert.notNull(database, "Database must not be null.");

        try {

            DatabaseDto databaseReqDto = buildDatabaseFromCatalogDatabase(databaseName, database);
            databaseServiceImpl.createDatabase(databaseReqDto);
        } catch (BusinessException e) {
            if (e.getCode().equals(DATABASE_EXISTS.getCode())) {
                if (!ignoreIfExists) {
                    throw new DatabaseAlreadyExistException(getName(), databaseName);
                }
            } else {
                throw new CatalogException(String.format("Failed to create database %s", databaseName), e);
            }
        } catch (Exception e) {
            throw new CatalogException(String.format("Failed to create database %s", databaseName), e);
        }

    }

    private DatabaseDto buildDatabaseFromCatalogDatabase(String databaseName, CatalogDatabase database) {
        DatabaseDto databaseReqDto = new DatabaseDto();
        databaseReqDto.setProperties(database.getProperties());
        databaseReqDto.setDatabaseName(databaseName);
        databaseReqDto.setComment(database.getComment());
        return databaseReqDto;
    }

    @Override
    public void dropDatabase(String databaseName, boolean ignoreIfNotExists, boolean cascade) throws DatabaseNotExistException, DatabaseNotEmptyException, CatalogException {


        Assert.notBlank(databaseName, DATABASE_NAME_REQUIRED.getMsg());
        try {

            databaseServiceImpl.deleteDatabase(databaseName, cascade);
        } catch (BusinessException e) {
            if (e.getCode().equals(DATABASE_NOT_EXISTS.getCode())) {
                if (!ignoreIfNotExists) {
                    throw new DatabaseNotExistException(getName(), databaseName);
                }
            } else if (e.getCode().equals(DATABASE_HAS_TABLE.getCode())) {
                throw new DatabaseNotEmptyException(getName(), databaseName);
            } else {
                throw new CatalogException(String.format("Failed to drop database %s", databaseName), e);
            }
        } catch (Exception e) {
            throw new CatalogException(String.format("Failed to drop database %s", databaseName), e);
        }

    }

    @Override
    public void alterDatabase(String databaseName, CatalogDatabase newDatabase, boolean ignoreIfNotExists) throws
            DatabaseNotExistException, CatalogException {

        Assert.notBlank(databaseName, DATABASE_NAME_REQUIRED.getMsg());

        DatabaseDto databaseDto = buildDatabaseFromCatalogDatabase(databaseName, newDatabase);
        try {
            databaseServiceImpl.updateDatabase(databaseDto);
        } catch (BusinessException e) {
            if (e.getCode().equals(DATABASE_NOT_EXISTS.getCode())) {
                if (!ignoreIfNotExists) {
                    throw new DatabaseNotExistException(getName(), databaseName);
                }
            } else {
                throw new CatalogException(String.format("Failed to alter database %s", databaseName), e);
            }
        } catch (Exception e) {
            throw new CatalogException(String.format("Failed to alter database %s", databaseName), e);
        }

    }

    @Override
    public List<String> listTables(String databaseName) throws DatabaseNotExistException, CatalogException {

        Assert.notBlank(databaseName, DATABASE_NAME_REQUIRED.getMsg());
        if (!databaseExists(databaseName)) {
            throw new DatabaseNotExistException(getName(), databaseName);
        }
        return tableServiceImpl.listTableNames(databaseName);
    }

    @Override
    public List<String> listViews(String databaseName) throws DatabaseNotExistException, CatalogException {

        //TODO 支持视图
        return Collections.emptyList();
    }

    @Override
    public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException {

        TableDto tableDto = tableServiceImpl.getTable( tablePath.getDatabaseName(), tablePath.getObjectName());
        if (tableDto == null) {
            throw new TableNotExistException(getName(), tablePath);
        }

        List<String> primaryKeys = new ArrayList<>(tableDto.getColumns().size());
        Schema.Builder schemaBuilder = Schema.newBuilder();
        for (ColumnDto column : tableDto.getColumns()) {
            AbstractDataType<DataType> dataType = DataTypeMapper.mappingTypes(column);
            schemaBuilder.column(column.getColumnName(), dataType);
            if (Boolean.TRUE.equals(column.getPrimaryKey())) {
                primaryKeys.add(column.getColumnName());
            }
        }
        schemaBuilder.primaryKey(primaryKeys);
        if (CollUtil.isNotEmpty(tableDto.getWatermarks())) {
            for (WatermarkDto watermark : tableDto.getWatermarks()) {
                schemaBuilder.watermark(watermark.getColumnName(), watermark.getExpression());
            }
        }
        Schema schema = schemaBuilder.build();
        return CatalogTable.of(schema, tableDto.getComment(), Collections.emptyList(), tableDto.getProperties());
    }


    @Override
    public boolean tableExists(ObjectPath tablePath) throws CatalogException {
        return tableServiceImpl.exists(tablePath.getDatabaseName(), tablePath.getObjectName());
    }

    @Override
    public void dropTable(ObjectPath tablePath, boolean ignoreIfNotExists) throws
            TableNotExistException, CatalogException {
        Assert.notNull(tablePath, TABLE_PATH_REQUIRED.getMsg());
        try {

            tableServiceImpl.deleteTable(tablePath.getDatabaseName(), tablePath.getObjectName());
        } catch (BusinessException e) {
            if (e.getCode().equals(TABLE_NOT_EXISTS.getCode())) {
                if (!ignoreIfNotExists) {
                    throw new TableNotExistException(getName(), tablePath);
                }
            } else {
                throw new CatalogException(String.format("Failed to drop table %s", tablePath.getFullName()), e);
            }
        } catch (Exception e) {
            throw new CatalogException(String.format("Failed to drop table %s", tablePath.getFullName()), e);
        }
    }

    @Override
    public void renameTable(ObjectPath tablePath, String newTableName, boolean ignoreIfNotExists) throws
            TableNotExistException, TableAlreadyExistException, CatalogException {

        Assert.notNull(tablePath, TABLE_PATH_REQUIRED.getMsg());
        Assert.notBlank(newTableName, TABLE_NAME_REQUIRED.getMsg());
        try {

            tableServiceImpl.renameTable(tablePath.getDatabaseName(), tablePath.getObjectName(), newTableName);
        } catch (BusinessException e) {
            if (e.getCode().equals(TABLE_NOT_EXISTS.getCode())) {
                if (!ignoreIfNotExists) {
                    throw new TableNotExistException(getName(), tablePath);
                }
            } else if (e.getCode().equals(TABLE_NAME_DUPLICATED.getCode())) {
                throw new TableAlreadyExistException(getName(), tablePath);
            } else {
                throw new CatalogException(String.format("Failed to rename table %s", tablePath.getFullName()), e);
            }
        } catch (Exception e) {
            throw new CatalogException(String.format("Failed to rename table %s", tablePath.getFullName()), e);
        }
    }

    @Override
    public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists) throws
            TableAlreadyExistException, DatabaseNotExistException, CatalogException {

        Assert.notNull(tablePath, TABLE_PATH_REQUIRED.getMsg());

        TableDto tableDto = buildTableFromCatalogTable(tablePath, table);
        try {
            tableServiceImpl.createTable(tableDto, ignoreIfExists);
        } catch (BusinessException e) {
            if (e.getCode().equals(DATABASE_NOT_EXISTS.getCode())) {
                throw new DatabaseNotExistException(getName(), tablePath.getDatabaseName());
            } else if (e.getCode().equals(TABLE_NAME_DUPLICATED.getCode())) {
                throw new TableAlreadyExistException(getName(), tablePath);
            } else {
                throw new CatalogException(String.format("Failed to create table %s", tablePath.getFullName()), e);
            }
        } catch (Exception e) {
            throw new CatalogException(String.format("Failed to create table %s", tablePath.getFullName()), e);
        }
    }

    private TableDto buildTableFromCatalogTable(ObjectPath tablePath, CatalogBaseTable catalogBaseTable) {


        TableDto tableDto = new TableDto();
        tableDto.setDatabaseName(tablePath.getDatabaseName());
        tableDto.setTableName(tablePath.getObjectName());
        tableDto.setComment(catalogBaseTable.getComment());
        tableDto.setProperties(catalogBaseTable.getOptions());

        Schema schema = catalogBaseTable.getUnresolvedSchema();
        //主键字段拼接起来
        Set<String> primaryKeys = schema.getPrimaryKey()
                .map(e -> new HashSet<>(e.getColumnNames()))
                .orElse(new HashSet<>());
        List<ColumnDto> columns = new ArrayList<>(schema.getColumns().size());
        List<WatermarkDto> watermarks = new ArrayList<>(schema.getWatermarkSpecs().size());
        for (Schema.UnresolvedColumn unresolvedColumn : schema.getColumns()) {

            if (unresolvedColumn instanceof Schema.UnresolvedPhysicalColumn) {
                AbstractDataType<?> dataType = ((Schema.UnresolvedPhysicalColumn) unresolvedColumn).getDataType();
                ColumnDto columnDto = DataTypeMapper.mappingColumn(dataType);
                columnDto.setColumnName(unresolvedColumn.getName());
                columnDto.setComment(unresolvedColumn.getComment().orElse(null));
                columnDto.setPrimaryKey(primaryKeys.contains(unresolvedColumn.getName()));
                columns.add(columnDto);
            } else {
                throw new CatalogException(String.format("Unsupported column type: %s", unresolvedColumn.getName()));
            }

        }
        for (Schema.UnresolvedWatermarkSpec watermarkSpec : schema.getWatermarkSpecs()) {

            WatermarkDto watermarkDto = new WatermarkDto();
            watermarkDto.setColumnName(watermarkSpec.getColumnName());

            Expression watermarkExpression = watermarkSpec.getWatermarkExpression();
            if (!(watermarkExpression instanceof SqlCallExpression)) {
                throw new CatalogException(String.format("Unsupported watermark type: %s", watermarkExpression.getClass()));
            }
            watermarkDto.setExpression(((SqlCallExpression) watermarkExpression).getSqlExpression());
            watermarks.add(watermarkDto);
        }
        tableDto.setColumns(columns);
        tableDto.setWatermarks(watermarks);

        return tableDto;
    }


    @Override
    public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists) throws
            TableNotExistException, CatalogException {

        TableDto tableDto = buildTableFromCatalogTable(tablePath, newTable);
        try {

            tableServiceImpl.updateTable(tableDto);
        } catch (BusinessException e) {
            if (e.getCode().equals(TABLE_NOT_EXISTS.getCode())) {
                throw new TableNotExistException(getName(), tablePath);
            } else {
                throw new CatalogException(String.format("Failed to alter table: %s", tablePath.getFullName()), e);
            }
        } catch (Exception e) {
            throw new CatalogException(String.format("Failed to alter table: %s", tablePath.getFullName()), e);
        }
    }

    @Override
    public List<String> listFunctions(String databaseName) throws DatabaseNotExistException, CatalogException {
        Assert.notBlank(databaseName, DATABASE_NAME_REQUIRED.getMsg());
        if (!databaseExists(databaseName)) {
            throw new DatabaseNotExistException(getName(), databaseName);
        }
        return functionServiceImpl.listFunctionNames(databaseName);
    }

    @Override
    public CatalogFunction getFunction(ObjectPath functionPath) throws FunctionNotExistException, CatalogException {
        FunctionDto functionDto = functionServiceImpl.getFunction(functionPath.getDatabaseName(), functionPath.getObjectName());
        if (functionDto == null) {
            throw new FunctionNotExistException(getName(), functionPath);
        }

        return new DacatCatalogFunctionImpl(functionDto);
    }

    @Override
    public boolean functionExists(ObjectPath functionPath) throws CatalogException {
        return functionServiceImpl.exists(functionPath.getDatabaseName(), functionPath.getObjectName());
    }

    @Override
    public void createFunction(ObjectPath functionPath, CatalogFunction function, boolean ignoreIfExists) throws
            FunctionAlreadyExistException, DatabaseNotExistException, CatalogException {
        Assert.notNull(functionPath, FUNCTION_PATH_REQUIRED.getMsg());

        FunctionDto functionDto = buildFunctionDto(functionPath, function);
        try {
            functionServiceImpl.createFunction(functionDto, ignoreIfExists);
        } catch (BusinessException e) {
            if (e.getCode().equals(DATABASE_NOT_EXISTS.getCode())) {
                throw new DatabaseNotExistException(getName(), functionPath.getDatabaseName());
            } else if (e.getCode().equals(FUNCTION_NAME_DUPLICATED.getCode())) {
                throw new FunctionAlreadyExistException(getName(), functionPath);
            } else {
                throw new CatalogException(String.format("Failed to create function %s", functionPath.getFullName()), e);
            }
        } catch (Exception e) {
            throw new CatalogException(String.format("Failed to create function %s", functionPath.getFullName()), e);
        }
    }

    private FunctionDto buildFunctionDto(ObjectPath functionPath, CatalogFunction function) {

        FunctionDto functionDto = new FunctionDto();
        functionDto.setDatabaseName(functionPath.getDatabaseName());
        functionDto.setFunctionName(functionPath.getObjectName());
        functionDto.setDescription(function.getDescription().orElse(null));
        functionDto.setClassName(function.getClassName());
        functionDto.setFunctionLanguage(function.getFunctionLanguage().name());
        functionDto.setIsGeneric(function.isGeneric());
        functionDto.setDetailDescription(function.getDetailedDescription().orElse(null));
        return functionDto;

    }

    @Override
    public void alterFunction(ObjectPath functionPath, CatalogFunction newFunction, boolean ignoreIfNotExists) throws
            FunctionNotExistException, CatalogException {

        //修改函数信息
        FunctionDto functionDto = buildFunctionDto(functionPath, newFunction);
        try {
            functionServiceImpl.updateFunction(functionDto, ignoreIfNotExists);
        } catch (BusinessException e) {
            if (e.getCode().equals(FUNCTION_NOT_EXISTS.getCode())) {
                if (!ignoreIfNotExists) {
                    throw new FunctionNotExistException(getName(), functionPath);
                }
            } else {
                throw new CatalogException(String.format("Failed to alter function %s", functionPath.getFullName()), e);
            }
        } catch (Exception e) {
            throw new CatalogException(String.format("Failed to alter function %s", functionPath.getFullName()), e);
        }

    }

    @Override
    public void dropFunction(ObjectPath functionPath, boolean ignoreIfNotExists) throws
            FunctionNotExistException, CatalogException {
        Assert.notNull(functionPath, FUNCTION_PATH_REQUIRED.getMsg());
        try {

            functionServiceImpl.deleteFunction(functionPath.getDatabaseName(), functionPath.getObjectName());
        } catch (BusinessException e) {
            if (e.getCode().equals(FUNCTION_NOT_EXISTS.getCode())) {
                if (!ignoreIfNotExists) {
                    throw new FunctionNotExistException(getName(), functionPath);
                }
            } else {
                throw new CatalogException(String.format("Failed to drop function %s", functionPath.getFullName()), e);
            }
        } catch (Exception e) {
            throw new CatalogException(String.format("Failed to drop function %s", functionPath.getFullName()), e);
        }
    }


    @Override
    public List<CatalogPartitionSpec> listPartitions(ObjectPath tablePath) throws
            TableNotExistException, TableNotPartitionedException, CatalogException {
        throw new UnsupportedOperationException();
    }

    @Override
    public List<CatalogPartitionSpec> listPartitions(ObjectPath tablePath, CatalogPartitionSpec partitionSpec) throws
            TableNotExistException, TableNotPartitionedException, PartitionSpecInvalidException, CatalogException {
        throw new UnsupportedOperationException();
    }

    @Override
    public List<CatalogPartitionSpec> listPartitionsByFilter(ObjectPath tablePath, List<Expression> filters) throws
            TableNotExistException, TableNotPartitionedException, CatalogException {
        throw new UnsupportedOperationException();
    }

    @Override
    public CatalogPartition getPartition(ObjectPath tablePath, CatalogPartitionSpec partitionSpec) throws
            PartitionNotExistException, CatalogException {
        throw new UnsupportedOperationException();
    }

    @Override
    public boolean partitionExists(ObjectPath tablePath, CatalogPartitionSpec partitionSpec) throws
            CatalogException {
        throw new UnsupportedOperationException();
    }

    @Override
    public void createPartition(ObjectPath tablePath, CatalogPartitionSpec partitionSpec, CatalogPartition
            partition, boolean ignoreIfExists) throws
            TableNotExistException, TableNotPartitionedException, PartitionSpecInvalidException, PartitionAlreadyExistsException, CatalogException {
        throw new UnsupportedOperationException();
    }

    @Override
    public void dropPartition(ObjectPath tablePath, CatalogPartitionSpec partitionSpec, boolean ignoreIfNotExists) throws
            PartitionNotExistException, CatalogException {
        throw new UnsupportedOperationException();
    }

    @Override
    public void alterPartition(ObjectPath tablePath, CatalogPartitionSpec partitionSpec, CatalogPartition
            newPartition, boolean ignoreIfNotExists) throws PartitionNotExistException, CatalogException {
        throw new UnsupportedOperationException();
    }
    @Override
    public CatalogTableStatistics getTableStatistics(ObjectPath tablePath) throws
            TableNotExistException, CatalogException {
        return null;
    }

    @Override
    public CatalogColumnStatistics getTableColumnStatistics(ObjectPath tablePath) throws
            TableNotExistException, CatalogException {
        return null;
    }

    @Override
    public CatalogTableStatistics getPartitionStatistics(ObjectPath tablePath, CatalogPartitionSpec partitionSpec) throws
            PartitionNotExistException, CatalogException {
        return null;
    }

    @Override
    public CatalogColumnStatistics getPartitionColumnStatistics(ObjectPath tablePath, CatalogPartitionSpec
            partitionSpec) throws PartitionNotExistException, CatalogException {
        return null;
    }

    @Override
    public void alterTableStatistics(ObjectPath tablePath, CatalogTableStatistics tableStatistics,
                                     boolean ignoreIfNotExists) throws TableNotExistException, CatalogException {
        throw new UnsupportedOperationException();
    }

    @Override
    public void alterTableColumnStatistics(ObjectPath tablePath, CatalogColumnStatistics columnStatistics,
                                           boolean ignoreIfNotExists) throws TableNotExistException, CatalogException, TablePartitionedException {
        throw new UnsupportedOperationException();
    }

    @Override
    public void alterPartitionStatistics(ObjectPath tablePath, CatalogPartitionSpec
            partitionSpec, CatalogTableStatistics partitionStatistics, boolean ignoreIfNotExists) throws
            PartitionNotExistException, CatalogException {
        throw new UnsupportedOperationException();
    }

    @Override
    public void alterPartitionColumnStatistics(ObjectPath tablePath, CatalogPartitionSpec
            partitionSpec, CatalogColumnStatistics columnStatistics, boolean ignoreIfNotExists) throws
            PartitionNotExistException, CatalogException {
        throw new UnsupportedOperationException();
    }
}

DataTypeMapper

DataTypeMapper用于与Flink原生的类型系统转换

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
public class DataTypeMapper {


    private static final Pattern LOGICAL_TYPE_PATTERN = Pattern.compile("^([a-zA-Z]+)(\\(((\\d+)(,\\s*(\\d+))?)\\))?\\s*((NOT)?\\s*NULL)?$");


    private DataTypeMapper() {
    }

    public static AbstractDataType<DataType> mappingTypes(@NonNull ColumnDto column) {

        AbstractDataType<DataType> dataType;
        FlinkDataTypeEnum flinkDataType = EnumUtil.getByCode(FlinkDataTypeEnum.class, column.getColumnType());
        if (flinkDataType == null) {
            throw new IllegalArgumentException("current not support " + column.getColumnType());
        }

        switch (flinkDataType) {
            case INT:
            case INTEGER:
                dataType = DataTypes.INT();
                break;
            case STRING:
                dataType = DataTypes.STRING();
                break;
            case DECIMAL:
                dataType = DataTypes.DECIMAL(column.getPrecision(), column.getScale());
                break;
            case VARCHAR:
                if (column.getColumnLength() == Integer.MAX_VALUE) {
                    dataType = DataTypes.STRING();
                } else {
                    dataType = DataTypes.VARCHAR(column.getColumnLength());
                }
                break;
            case TIMESTAMP:
                dataType = DataTypes.TIMESTAMP();
                break;
            case BIGINT:
                dataType = DataTypes.BIGINT();
                break;
            case TINYINT:
                dataType = DataTypes.TINYINT();
                break;
            case DATE:
                dataType = DataTypes.DATE();
                break;
            default:
                throw new IllegalArgumentException("current not support " + column.getColumnType());
        }
        if (BooleanUtil.isFalse(column.getNullable())) {
            dataType = dataType.notNull();
        }
        return dataType;
    }

    public static ColumnDto mappingColumn(@NonNull AbstractDataType<?> dataType) {

        ColumnDto column = new ColumnDto();
        if (dataType instanceof AtomicDataType) {

            LogicalType logicalType = ((AtomicDataType) dataType).getLogicalType();

            Matcher matcher = LOGICAL_TYPE_PATTERN.matcher(logicalType.asSummaryString());
            if (!matcher.matches()) {
                throw new IllegalArgumentException("无法匹配的字段类型");
            }
            column.setColumnType(matcher.group(1));
            column.setNullable(matcher.group(8) == null);
            String lengthStr = matcher.group(4);
            if (lengthStr != null) {
                Integer length = Integer.valueOf(lengthStr);
                column.setColumnLength(length);
                column.setPrecision(length);
            }
            String scaleStr = matcher.group(6);
            if (scaleStr != null) {
                Integer scale = Integer.valueOf(scaleStr);
                column.setScale(scale);
            }
            column.setOriginalDefinition(logicalType.asSummaryString());
        }

        return column;
    }

}
FlinkDataTypeEnum

flink的数据类型枚举

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
@Getter
public enum FlinkDataTypeEnum implements IEnum.StrEnum {

    INT(Type.INT),
    INTEGER(Type.INTEGER),
    TINYINT(Type.TINYINT),
    VARCHAR(Type.VARCHAR),
    TIMESTAMP(Type.TIMESTAMP),
    STRING(Type.LONGVARCHAR),
    DECIMAL(Type.DECIMAL),
    BIGINT(Type.BIGINT),
    CHAR(Type.CHAR),
    SMALLINT(Type.SMALLINT),
    BOOLEAN(Type.BOOLEAN),
    BINARY(Type.BINARY),
    VARBINARY(Type.VARBINARY),
    FLOAT(Type.FLOAT),
    DATE(Type.DATE),
    TIME(Type.TIME),
    ;


    private final Type primaryType;
    private final Set<Type> types;


    FlinkDataTypeEnum(@NotEmpty Type... types) {
        this.primaryType = types[0];
        this.types = ImmutableSet.copyOf(types);
    }

    public static FlinkDataTypeEnum getByType(Type type) {

        for (FlinkDataTypeEnum flinkDataType : values()) {
            if (flinkDataType.types.contains(type)) {
                return flinkDataType;
            }
        }
        return null;
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31


public class MyJdbcCatalog extends JdbcCatalog {

    public MyJdbcCatalog(String catalogName, String defaultDatabase, String username, String password, String baseUrl) {
        super(catalogName, defaultDatabase, username, password, baseUrl);
    }

    @Override
    public void open() throws CatalogException {
        // 在打开Catalog时，我们可以在这里实现初始化操作
    }

    @Override
    public void close() throws CatalogException {
        // 在关闭Catalog时，我们可以在这里释放资源
    }

    @Override
    public List<String> listTables(String databaseName) throws CatalogException {
        // 重写listTables方法，实现自定义的表列表查询逻辑
    }

    @Override
    public CatalogTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException {
        // 重写getTable方法，实现自定义的表查询逻辑
    }

    // 其他重写方法...

}

单元测试

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
@SpringBootTest(classes = Application.class)
@Slf4j
class WritableJdbcCatalogTest {

    private TableEnvironment tableEnv;

    @Autowired
    private DatabaseServiceImpl databaseServiceImpl;
    @Autowired
    private TableServiceImpl tableServiceImpl;
    @Autowired
    private FlinkProperties flinkProperties;
    @Autowired
    private FunctionServiceImpl functionServiceImpl;

    @BeforeEach
    void setUp() throws ExecutionException, InterruptedException {
        EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        tableEnv = TableEnvironment.create(environmentSettings);

        String catalogName = "mysql-catalog";
        String databaseName = "test";
        Catalog writableMysqlCatalog = new WritableJdbcCatalog(flinkProperties, databaseServiceImpl, tableServiceImpl, functionServiceImpl);
        tableEnv.registerCatalog(catalogName, writableMysqlCatalog);
        tableEnv.useCatalog(catalogName);
        tableEnv.executeSql("CREATE DATABASE test;")
                .await();
        tableEnv.useDatabase(databaseName);
        tableEnv.executeSql("CREATE TABLE datagen (\n" +
                        " f_sequence INT,\n" +
                        " f_random INT,\n" +
                        " f_random_str STRING,\n" +
                        " f_random_varchar VARCHAR(20),\n" +
                        " update_time TIMESTAMP(3),\n" +
                        " primary key (f_sequence) NOT ENFORCED, \n" +
                        " WATERMARK FOR update_time AS update_time - INTERVAL '5' SECOND \n" +
                        ") WITH (\n" +
                        " 'connector' = 'datagen');")
                .await();
    }

    @AfterEach
    void tearDown() throws ExecutionException, InterruptedException {
        tableEnv.executeSql("DROP DATABASE IF EXISTS test CASCADE;")
                .await();
    }

    @Test
    void listDatabases() {

        String[] databases = tableEnv.listDatabases();
        Assertions.assertArrayEquals(new String[]{"test"}, databases);
    }

    @Test
    void getDatabase() {

        tableEnv.useDatabase("test");
        Assertions.assertDoesNotThrow(() -> new Exception());

    }


    @Test
    void createAndDropDatabase() throws ExecutionException, InterruptedException {

        tableEnv.executeSql("CREATE DATABASE test1").await();
        String[] databases = tableEnv.listDatabases();
        Assertions.assertArrayEquals(new String[]{"test", "test1"}, databases);

        tableEnv.executeSql("DROP DATABASE test1;").await();

        databases = tableEnv.listDatabases();
        Assertions.assertArrayEquals(new String[]{"test"}, databases);

    }

    @Test
    void listTables() {

        String[] tables = tableEnv.listTables();
        Assertions.assertArrayEquals(new String[]{"datagen"}, tables);

    }

    @Test
    void selectTableData() throws Exception {

        TableResult tableResult = tableEnv.executeSql("select * from datagen limit 3;");
        try (CloseableIterator<Row> rows = tableResult.collect()) {
            for (int i = 0; i < 3; i++) {

                Row row = rows.next();
                log.info("received row: {}", row);
            }
        }
        tableResult.getJobClient().ifPresent(JobClient::cancel);
    }

    @Test
    void dropTable() throws ExecutionException, InterruptedException {

        String[] tables = tableEnv.listTables();
        Assertions.assertArrayEquals(new String[]{"datagen"}, tables);
        tableEnv.executeSql("DROP TABLE datagen;")
                .await();
        tables = tableEnv.listTables();
        Assertions.assertArrayEquals(new String[]{}, tables);
    }

    @Test
    void renameTable() throws ExecutionException, InterruptedException {

        tableEnv.executeSql("ALTER TABLE datagen RENAME TO datagen1;")
                .await();
        String[] tables = tableEnv.listTables();
        Assertions.assertArrayEquals(new String[]{"datagen1"}, tables);
        tableEnv.executeSql("ALTER TABLE datagen1 RENAME TO datagen;")
                .await();
        tables = tableEnv.listTables();
        Assertions.assertArrayEquals(new String[]{"datagen"}, tables);
    }

    @Test
    void testFunction() throws Exception {


        tableEnv.createFunction("c_split", SplitFunction.class, true);
        try (CloseableIterator<Row> iterator = tableEnv.executeSql("" +
                        "select word, length" +
                        " from LATERAL TABLE(c_split('a,b,cd',',')) limit 3;")
                .collect();) {
            List<Row> rows = new ArrayList<>();
            iterator.forEachRemaining(rows::add);

            Assertions.assertEquals(3, rows.size());
            Assertions.assertEquals("a", rows.get(0).getField(0));
            Assertions.assertEquals(1, rows.get(0).getField(1));
            Assertions.assertEquals("b", rows.get(1).getField(0));
            Assertions.assertEquals(1, rows.get(1).getField(1));
            Assertions.assertEquals("cd", rows.get(2).getField(0));
            Assertions.assertEquals(2, rows.get(2).getField(1));

        }

    }
}

其中的SplitFunction为flink官网上的示例,增加了一个seperator参数

1
2
3
4
5
6
7
8
9
10
11
@FunctionHint(output = @DataTypeHint("ROW<word STRING, length INT>"))
public class SplitFunction extends TableFunction<Row> {

    public void eval(String str,String separator) {
        for (String s : str.split(separator)) {
            // use collect(...) to emit a row
            collect(Row.of(s, s.length()));
        }
    }

}

总结

本文介绍了如何使用基于jdbc的自定义Catalog，在FlinkSQL中实现自定义的数据源和表管理。自定义Catalog可以更灵活地管理数据源和表，满足不同场景下的需求。通过实现自定义Catalog，我们可以更加灵活地管理数据源和表，满足不同场景下的需求。

后续工作

本文的自定义Catalog只支持在springboot环境中使用,后续如果我们借助JAVA的SPI机制,实现了Flink的CatalogFactorty后,也可以让自定义的Catalog在交互式命令行中使用,但是这样必须考虑MybatisPlus的适配问题,比较麻烦,而且实际上只有调试的时候能够用到,后续有机会的话再来填这个坑= =

大数据
#java #flink
FlinkSQL-基于jdbc的自定义Catalog
http://example.com/2023/03/19/flink/FlinkSQL-基于jdbc的自定义Catalog/
作者
敬文
发布于
2023年3月19日
许可协议