waf-calcite-动态数据管理

calcite官网：http://calcite.apache.org/

Apache Calcite是一款开源的动态数据管理框架，它提供了标准的 SQL 语言、多种查询优化和连接各种数据源的能力，但不包括数据存储、处理数据的算法和存储元数据的存储库。

这里基于Apache Calcite做为基础组件进行封装，提供一整套数据源管理与使用的接口抽象组件，通过进行接口注入就可以跟使用本地service一样调用。

1.2

2025-11-26 发布1.2.0-SNAPSHOT快照版本发布到nexus私服

功能列表

• 新增数据源元数据相关接口：支持获取物理表列表、生成建表语句、SQL 验证与解析、物理表详情查询等
• 新增索引查询接口；列信息接口扩展字段缓冲区长度、作用域目录等属性，并支持外键信息查看
• 扩展 Calcite Server 语法，支持解析建表语句中表级 COLLATE 属性

依赖升级

• Apache Calcite升级为1.41.0

1.1

2025-11-03 发布1.1.1正式版本发布到nexus私服

2025-07-17 发布1.1.0正式版本发布到nexus私服

功能列表

• 1、支持开启calcite-server模块（DDL支持），并支持通过环境变量控制开关
• 2、Calcite不兼容UTF8字符集问题修复
• 3、Calcite生成SQL语句时会自动给中文加上字符集前缀导致具体数据源执行异常问题修复
• 4、重构 Calcite 连接方式，改为按需加载元数据，避免全量加载元数据导致首次查询耗时较长问题
• 5、抽象 Calcite 初始化逻辑，支持自定义扩展；优化数据源注册性能，改为并行处理
• 6、优化 Calcite 数据源注册性能（10s+优化至1-2s左右）

依赖升级

• Apache Calcite升级为1.40.0

1.0

2025-03-05 发布1.0.0正式版本发布到nexus私服（当前在试用的应用：数创平台、对象视角、指标平台）

2024-11-26 发布1.0.1-SNAPSHOT快照版本发布到nexus私服

功能列表

• 1、提供数据源前后端配置管理功能（Vue2版本合并waf-web v3.x分支；Vue3版本还在开发中）
• 2、当前适配完成的数据源：Mysql、Oracle、DB2、PostgreSQL、clickhouse、huaweicloud.dws、ArgonDB、Kingbase、hive2
• 3、适配指标数据源（数据中台指标平台接口）
• 4、抽象数据源管理、calcite数据源、指标数据源接口，可通过配置自定义实现
• 5、支持数据库驱动为本地路径设置
• 6、高版本druid连接数据源失败后一直重连问题修复
• 7、数据源支持配置集群参数（如：clickhouse集群模式）

依赖升级

• waf-parent 更新到3.2.5
• Apache Calcite 更新到1.38.0

更新步骤

• 在后端pom中添加依赖，当前版本最新为1.2.0-SNAPSHOT

#=========================后端更新==============================#
# 将maven waf-calcite版本改为 1.2.0-SNAPSHOT
<waf-calcite.version>1.2.0-SNAPSHOT</waf-calcite.version>

<dependency>
	<groupId>com.wiseda.waf</groupId>
	<artifactId>waf-calcite-core</artifactId>
	<version>${waf-calcite.version}</version>
</dependency>
#=========================后端更新==============================#

1、数据库脚本

这里提供一份数据源的配置存储表，业务系统可以参照扩展数据库表结构与接口

-- 数据源表结构
CREATE TABLE sys_datasource
(
    id          varchar(40) NOT NULL COMMENT '数据源id',
    name        varchar(50) NOT NULL COMMENT '数据源名称',
    type        varchar(50) NOT NULL COMMENT '数据源类型：mysql、db2、API',
    parent_id   varchar(40) NOT NULL COMMENT '父节点id',
    config      text COMMENT '数据源配置',
    status      tinyint(1) NOT NULL COMMENT '1:启用;0:停用',
    sort_num    int(11) DEFAULT NULL COMMENT '排序号',
    remark      varchar(300) DEFAULT NULL COMMENT '描述',
    create_by   varchar(40)  DEFAULT NULL COMMENT '创建者',
    create_time datetime     DEFAULT NULL COMMENT '创建时间',
    update_by   varchar(40)  DEFAULT NULL COMMENT '更新者',
    update_time datetime     DEFAULT NULL COMMENT '更新时间',
    PRIMARY KEY (id),
    KEY idx_parent_id (parent_id) USING BTREE
) COMMENT='数据源管理';

-- 数据管理管理菜单 挂入 系统配置 菜单下
INSERT INTO sys_permission (id, name, parent_id, `type`, icon, url, component, perms, is_route, is_use, is_visible, is_search, is_cache, sort_num, remark, is_deleted, micro_app, micro_app_component, create_by, create_time, update_by, update_time) VALUES('1856999090590687232', '数据源管理', 'ac62ef18339d48c0b0e3b13a31a54ac7', '0', 'nested', '/datasource/sysSourceCon', 'datasource/SysSourceCon', 'Datasource:View', NULL, 1, 1, NULL, 1, 30, '', NULL, '', '', 'admin', '2024-11-18 14:35:53', 'admin', '2024-11-18 17:05:14');
INSERT INTO sys_permission (id, name, parent_id, `type`, icon, url, component, perms, is_route, is_use, is_visible, is_search, is_cache, sort_num, remark, is_deleted, micro_app, micro_app_component, create_by, create_time, update_by, update_time) VALUES('1856999090590687233', '新增', '1856999090590687232', '1', '', '', '', 'Datasource:Add', NULL, 1, 1, NULL, NULL, 10, '', NULL, NULL, NULL, 'admin', '2024-11-18 14:35:53', NULL, NULL);
INSERT INTO sys_permission (id, name, parent_id, `type`, icon, url, component, perms, is_route, is_use, is_visible, is_search, is_cache, sort_num, remark, is_deleted, micro_app, micro_app_component, create_by, create_time, update_by, update_time) VALUES('1856999090590687234', '修改', '1856999090590687232', '1', '', '', '', 'Datasource:Update', NULL, 1, 1, NULL, NULL, 20, '', NULL, NULL, NULL, 'admin', '2024-11-18 14:35:53', NULL, NULL);
INSERT INTO sys_permission (id, name, parent_id, `type`, icon, url, component, perms, is_route, is_use, is_visible, is_search, is_cache, sort_num, remark, is_deleted, micro_app, micro_app_component, create_by, create_time, update_by, update_time) VALUES('1856999090594881535', '删除', '1856999090590687232', '1', '', '', '', 'Datasource:Del', NULL, 1, 1, NULL, NULL, 30, '', NULL, NULL, NULL, 'admin', '2024-11-18 14:35:53', NULL, NULL);

2、后端应用

后端主要在pom.xml里引用相应版本，直接调用接口就可以使用，目前最新版本：1.2.0-SNAPSHOT

• pom.xml配置

<waf.calcite.version>1.2.0-SNAPSHOT</waf.calcite.version>

 <!--calcite 核心包引用-->
<dependency>
    <groupId>com.wiseda.waf</groupId>
    <artifactId>waf-calcite-core</artifactId>
    <version>${waf.calcite.version}</version>
</dependency>

• yaml配置

waf:
  server:
    calcite:
      # 是否初始化Calcite连接（默认启用，加载数据源)，1.1版本后属性名由enabled修改为 => init-on-startup
      init-on-startup: ${WAF_CALCITE_INIT:true}
      # 是否分布式部署（多实例部署时，默认开启）
      distributed: ${WAF_CALCITE_DISTRIBUTED:true}    
      # 是否开启服务器模块支持（支持DDL语句，默认关闭）
      server-module: ${WAF_CALCITE_SERVER:false}
      # 自定义JDBC驱动路径
      driver-path: ${WAF_CALCITE_DRIVER_PATH:}
      # 定义接口模式,默认LOCAL（LOCAL-本地库;CUSTOM-自定义实现）
      interface-list:
        # 接口列表（sys-datasource-conf:数据源配置;calcite-mgr:Calcite数据源管理;calcite-provider:Calcite动态数据源相关api提供;jdbc-datasource:JDBC数据源;metric-datasource:指标数据源;metric-dmm-datasource:数据中台指标数据源)
        sys-datasource-conf: ${WAF_CALCITE_SYS_DATASOURCE_CONF:LOCAL}
        calcite-mgr: ${WAF_CALCITE_CALCITE_MGR:LOCAL}
        calcite-provider: ${WAF_CALCITE_CALCITE_PROVIDER:LOCAL}
        jdbc-datasource: ${WAF_CALCITE_JDBC_DATASOURCE:LOCAL}
        metric-datasource: ${WAF_CALCITE_METRIC_DATASOURCE:LOCAL}
        metric-dmm-datasource: ${WAF_CALCITE_METRIC_DMM_DATASOURCE:LOCAL}

init-on-startup：是否随应用启动时启用（如果设置为false，需自行加载数据源到calcite）

distributed：如果只部署单实例或1个pod，可以设置为false；默认支持分布式多实例（多pod）部署，即在数据源的启停操作时会多个实例同步状态。

server-module：开启后使用Calcite解析/验证SQL等接口将支持DDL语句。

分布式需要在主应用的RedisConfig进行队列订阅

// RedisConfig 配置中添加订阅队列
    @Bean
    public RedisMessageListenerContainer listenerContainer(RedisConnectionFactory connectionFactory) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        // 订阅消息主题
        container.addMessageListener(new CalciteMessageListener(), 
                                     new ChannelTopic(IDatasourceConstant.SYS_CALCITE_QUEUE));
        // container 可以添加多个 messageListener
        return container;
    }

driver-path：可能通过配置驱动的路径，来加载所需的版本。适合于独立部署在服务器，如通过jar包的方式；对于容器的方式，可以在构建镜像时把jar包一并打进去或者通过pom中直接引用。也可以在应用部署后，通过数据卷挂载到容器中来，总之通过路径的方式需要能访问到具体路径下的jar包驱动。

interface-list：接口列表，waf-calcite提供一套默认的接口实现，同时可以自定义去实现里面任意接口或接口方法。

3、前端应用

前端应用可以与waf-web:v3.x分支进行合并，可以手工复制下面相关文件到自身应用，参照git log中2024-11-18 feat: 新增数据源管理模块提交

• src/api/admin/sysDatasource.js
• src/utils/datasource.js
• src/icons/svg
• src/assets/fontIcons、src/assets/fontIcons-cs
• src/views/datasource

常用服务介绍

• SysDatasourceConfService：系统数据源配置服务（sys_datasource表相关服务）

  方法名	描述
  getDatasourceCache	获取数据源信息
  getDatasourceList	获取数据源列表
  addDatasource	新增数据源到sys_datasource，并且注册到Calcite中。
  updateDatasource	修改sys_datasource中的数据源信息，同时更新到Calcite中。
  ....

• CalciteDatasourceService：对注册到Calcite中的数据源提供的一系列函数。

  方法名	描述
  getDatabase	获取指定数据源中所有的database
  getSchemas	获取指定数据源中所有的schema
  getTables	获取数据源中所有物理表或视图
  getColumns	获取物理表中所有列
  getJdbcResultList	执行单数据源的查询语句
  getCalciteResultList	执行跨数据源的查询语句
  executeUpdate	执行给定的SQL语句（包括DML和DDL）
  generatorCreateTableSql	根据数据源id、物理表名生成建表语句
  validateSql	SQL校验，支持多行SQL同时校验（支持DDL语句，但暂不完善）
  parserSql	解析SQL语句，返回SQL语法树
  ...

• CalciteManagerService：管理Calcite的相关函数。

  方法名	描述
  initCalciteConnection	初始化Calcite连接
  getConnection	获取Calcite连接
  getRootSchema	获取Calcite根模式
  getDataSource	获取注册到Calcite中的数据源
  getSQLDialect	获取数据源对应的SQL方言实例
  getConnectionPool	获取注册到calcite中的数据源（从连接池获取）
  addSchema	添加schema(数据源)到calcite中
  updateSchemaToCalcite	更新schema(数据源)到calcite中，不存在将新增数据源
  removeCalciteSchema	移除注册到calcite中的数据源
  ...

集成 waf-calcite 组件示例

使用步骤：

1. 初始化数据源到Calcite中（通过提供的数据源页面添加）
2. 从waf-calcite中选择需要用到的bean，然后注入到类中即可使用。

示例1、跨库查询能力

项目背景：[数创平台]需支持跨多源的统一语义层查询，通过 waf-calcite 组件进行数据源的统一管理，并且实现跨库查询能力。

首先需要准备A、B两个数据源，并在数据源中执行相应脚本：

# 在数据源A中创建班级表
CREATE TABLE `t_class_info` (
  `id` int NOT NULL COMMENT '班级id',
  `name` varchar(100) COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '班级名称',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
INSERT INTO t_class_info (id, name) VALUES(1, '1班');
INSERT INTO t_class_info (id, name) VALUES(2, '2班');

# 在数据源B中创建学生表
CREATE TABLE `t_student_info` (
  `id` int NOT NULL COMMENT '学生id',
  `name` varchar(100) COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '学生姓名',
  `class_id` int NOT NULL COMMENT '所属班级',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='学生表';
INSERT INTO t_student_info (id, name, class_id) VALUES(1, '张三', 1);
INSERT INTO t_student_info (id, name, class_id) VALUES(2, '李四', 1);
INSERT INTO t_student_info (id, name, class_id) VALUES(3, '王五', 2);

假设已注册以下数据源：

• 数据源A注册到sys_datasource表后id为：123
• 数据源B注册到sys_datasource表后id为：456

使用步骤：

1. 组装查询语句（在跨库环境下需要给查询SQL的表名增加限定名：s_a_ + 数据源id）

   # 例：
   # 在同数据库下SQL语句为：
   select 
   	si.name as sName, c.name as className
   from t_class_info c, t_student_info si
   where c.id  = si.class_id 
   
   # 在跨库环境下应为：
   select 
   	si.name as sName, c.name as className
   from s_a_123.t_class_info c, s_a_456.t_student_info si
   where c.id  = si.class_id 
   

2. 注入 CalciteDatasourceService 服务，并调用getCalciteResultList函数传入sql即可。

   @Autowired
   private CalciteDatasourceService service;
   @Autowired
   private SysDatasourceConfService datasourceService;
   
   public void execCalciteSql(String sql) throws WafCommonException {
       Datasource sourceA = datasourceService.getDatasourceCache("123");
       Datasource sourceB = datasourceService.getDatasourceCache("456");
       Map<String, Datasource> dsMap = Map.of("123", sourceA, "456", sourceB);
       // 跨表查询的结果集封装在TableResultDTO中
       TableResultDTO resultList = service.getCalciteResultList(sql, dsMap);
   }
   

示例2、SQL校验能力

项目背景：在「大数据平台」中使用，加强数据治理与操作安全性，对用户提交的 DML 和 DDL语句实施细粒度权限与合规性校验。 基于 Calcite 的 SQL 解析与验证能力，平台扩展实现表级/列级访问控制，在语句执行前完成静态分析与风险拦截。

依旧拿示例1中的数据源信息、物理表信息举例：

• 注入 CalciteDatasourceService 服务，调用validateSql函数即可完成验证。

@Autowired
private CalciteDatasourceService service;

public void sqlValidator1() {
    String datasourceId = "123";
    StringBuilder validSql = new StringBuilder();
    validSql.append("select * from t_class_info;");
    validSql.append("-- 测试注释\n");
    validSql.append("select c from t_class_info;");
    try {
    	service.validateSql(datasourceId, validSql);
    } catch (CustomCalciteContextException e) {
        // 异常信息：From line 1, column 28 to line 3, column 8: Column 'c' not found in any table: Column 'c' not found in any table
        System.out.println("异常消息：" + e.getMessage());
         // 异常类：CustomCalciteContextException，可通过获取异常中的属性抛出自定义异常（开始行、结束行等）。
        String message = String.format("第%s行, %s到第%s行，第%s列发生异常，异常对象：%s", e.getPosLine(), e.getPosColumn(), e.getEndPosLine(), e.getEndPosColumn(), e.getNode().toString());
        System.out.println(message);
    }
}