Apache Calcite动态SQL元数据管理实践：DDL操作、SqlParser解析、SqlValidator校验与UDF自定义

引言

在之前的实践中，大量篇幅介绍了如何通过实现 Calcite 的自定义表元数据来实践 SQL 查询数据的功能。元数据的定义就像一个驱动器，桥接了 Calcite SQL 与存储介质之间的关系。但就目前实现结果来看，我们仍然可以思考一些可以优化的方向。

本文将以一个 CSV 查询的示例代码为基础，进行优化和功能扩展。

优化方向与实践

这里的优化方向偏向于在实践过程中还可以继续思考、补充的地方。

1. 动态管理元数据：去掉 model.json 文件引用

在基础的 CSV 查询示例中，使用了 model.json 文件来静态定义元数据。这种方式在实际应用中并不方便。更优的做法是使用 RootSchema 动态添加元数据，从而摆脱对配置文件依赖。这一过程涉及到对Java和数据库/中间件层的动态管理。

YzhouCsvTest_withoutjson.java

public class YzhouCsvTest_withoutjson {
  public static void main(String[] args) throws SQLException {
    Properties props = new Properties();
    props.setProperty("caseSensitive", "false");
    props.setProperty("lex", "JAVA");
    try (Connection connection = DriverManager.getConnection("jdbc:calcite:", props);
         CalciteConnection calciteConnection = connection.unwrap(CalciteConnection.class)) {
      SchemaPlus rootSchema = calciteConnection.getRootSchema();
      File csvDir = new File("javamain-calcite\\src\\main\\resources\\sales");
      CsvSchema csvSchema = new CsvSchema(csvDir, CsvTable.Flavor.SCANNABLE);
      // 动态添加表到模式
      rootSchema.add("depts", csvSchema.getTable("DEPTS"));
      rootSchema.add("sdepts", csvSchema.getTable("SDEPTS"));
      String sql = "select deptno,name from depts";
      try (Statement statement = connection.createStatement();
           ResultSet resultSet = statement.executeQuery(sql)) {
        print(resultSet);
      }
    }
  }

  private static void print(ResultSet resultSet) throws SQLException {
    final ResultSetMetaData metaData = resultSet.getMetaData();
    final int columnCount = metaData.getColumnCount();
    while (resultSet.next()) {
      for (int i = 1; ; i++) {
        System.out.print(resultSet.getString(i));
        if (i < columnCount) {
          System.out.print(", ");
        } else {
          System.out.println();
          break;
        }
      }
    }
  }
}

执行结果：

10, Sales
20, Marketing
30, Accounts

2. 实现动态DDL：通过 SQL 语句操作元数据

无论是 model.json 文件还是自定义表元数据，在程序启动时就决定了元数据的结构，这导致后期难以扩展。我们可以参考 Flink SQL 中 Catalog 的思路，实现类似 MySQL 的 DDL 功能，通过 SQL 语句动态地对元数据进行创建、修改等操作。

DdlExample.java

public class DdlExample {
  public static void main(String[] args) throws SQLException {
    // 1. 配置属性，指定支持 DDL 的解析器
    Properties props = new Properties();
    props.setProperty("lex", "JAVA");
    props.put(
            CalciteConnectionProperty.PARSER_FACTORY.camelName(),
            ServerDdlExecutor.class.getName() + "#PARSER_FACTORY"
    );
    // 2. 创建连接
    try (Connection conn = DriverManager.getConnection("jdbc:calcite:", props);
         Statement stmt = conn.createStatement()) {
      // === 用 SQL 来定义元数据结构 ===
      stmt.execute("CREATE SCHEMA sales");
      stmt.execute("CREATE TABLE sales.depts(deptno int, name varchar(20))");
      stmt.execute("INSERT INTO sales.depts VALUES (1, 'IT'), (2, 'HR')");
      // === 查询数据 ===
      try (ResultSet rs = stmt.executeQuery("SELECT * FROM sales.depts")) {
        print(rs);
      }
    }
  }

  private static void print(ResultSet rs) throws SQLException {
    ResultSetMetaData meta = rs.getMetaData();
    int cols = meta.getColumnCount();
    while (rs.next()) {
      for (int i = 1; i <= cols; i++) {
        System.out.print(rs.getString(i));
        if (i < cols) System.out.print(", ");
      }
      System.out.println();
    }
  }
}

执行结果：

1, IT
2, HR

3. 集成自定义 UDF：结合 SqlParser 与 SqlValidator

自定义 UDF 在大数据领域的数据库引擎中非常常见，Calcite 也支持此功能，但其定义和执行方式有自己的生命周期。理解 Calcite SQL 的处理流程（解析、校验、转换、执行）对于正确集成 UDF 至关重要。

在基础的 CSV 查询示例中，我们执行 select deptno,name from depts。现在我们需要添加一个 UDF 函数 addStr，它接收一个字符串参数，并为其添加固定的后缀，例如：输入 "yzhou"，输出 "yzhou_addStr"。

3.1 尝试一：通过继承 SqlFunction 定义 UDF

首先尝试通过继承 SqlFunction 并注册到自定义的 SqlOperatorTable 来定义 UDF。这种方式负责定义语法和类型校验规则。

UAddStr.java

public class UAddStr extends SqlFunction {
  public UAddStr(String name, SqlKind kind, @Nullable SqlReturnTypeInference returnTypeInference, @Nullable SqlOperandTypeInference operandTypeInference, @Nullable SqlOperandTypeChecker operandTypeChecker, SqlFunctionCategory category) {
    super(name, kind, returnTypeInference, operandTypeInference, operandTypeChecker, category);
  }
}

ExtendedSqlOperatorTable.java

public class ExtendedSqlOperatorTable extends ReflectiveSqlOperatorTable {
  private static ExtendedSqlOperatorTable instance;
  public static synchronized ExtendedSqlOperatorTable instance() {
    if (instance == null) {
      instance = new ExtendedSqlOperatorTable();
      instance.init();
    }
    instance.register(addStr());
    return instance;
  }
  private static SqlFunction addStr() {
    return new UAddStr("addStr",
            SqlKind.OTHER_FUNCTION,
            ReturnTypes.VARCHAR,
            null,
            OperandTypes.STRING,
            SqlFunctionCategory.USER_DEFINED_FUNCTION);
  }
}

接下来，在查询示例中集成 SQL 解析 (SqlParser) 和校验 (SqlValidator) 功能。需要注意的是，SqlParser.parseQuery() 仅进行语法解析，不会校验字段、表名或 UDF 是否存在。真正的校验在 SqlValidator.validate() 阶段完成。

YzhouCsvTest_withvalidator_withsqlfunction.java (核心片段)

private static void validateSql(Properties props,String sql, SchemaPlus rootSchema) throws SqlParseException {
    // 解析 SQL
    SqlParser.Config config = SqlParser.configBuilder()
            .setCaseSensitive(false)
            .setLex(Lex.JAVA)
            .build();
    SqlParser parser = SqlParser.create(sql, config);
    SqlNode sqlNodeParsed = parser.parseQuery();
    System.out.println("[parsed sqlNode]");
    System.out.println(sqlNodeParsed);
    // 校验 SQL
    JavaTypeFactoryImpl sqlTypeFactory = new JavaTypeFactoryImpl();
    CalciteCatalogReader catalogReader = new CalciteCatalogReader(
            CalciteSchema.from(rootSchema),
            CalciteSchema.from(rootSchema).path(null),
            sqlTypeFactory,
            new CalciteConnectionConfigImpl(props));
    // 使用自定义的 OperatorTable 进行校验
    SqlValidator validator = SqlValidatorUtil.newValidator(
            ExtendedSqlOperatorTable.instance(),
            catalogReader,
            sqlTypeFactory,
            SqlValidator.Config.DEFAULT);
    SqlNode sqlNodeValidated = validator.validate(sqlNodeParsed);
    System.out.println("\n[validated sqlNode]");
    System.out.println(sqlNodeValidated);
}

执行上述代码，解析和校验阶段能成功通过，但在最终执行查询时会报错：No match found for function signature addstr(<CHARACTER>)。这是因为通过 SqlFunction 定义的 UDF 仅参与了语法和类型的声明与校验，并未与具体的执行逻辑绑定。Calcite 在执行阶段无法找到该函数的实现。

3.2 尝试二：通过 Schema 注册 UDF 及其实现

第二种方式是将包含实际执行逻辑的 Java 方法注册到 Schema 中。这是更常见的 UDF 定义方式。

UAddStr.java (修订版)

public class UAddStr {
  public static String addStr(String str) {
    return str + "_addStr";
  }
}

YzhouCsvTest_withvalidator_withschema.java (核心片段)

public static void main(String[] args) throws SQLException {
    ...
    SchemaPlus rootSchema = calciteConnection.getRootSchema();
    ...
    // 通过 ScalarFunctionImpl 将 Java 方法注册为 UDF
    rootSchema.add("addstr", ScalarFunctionImpl.create(UAddStr.class, "addStr"));
    String sql = "select deptno,addstr(name) from depts";
    // 如果注释掉校验，直接执行，UDF 工作正常
    // validateSql(props,sql,rootSchema);
    try (Statement statement = connection.createStatement();
         ResultSet resultSet = statement.executeQuery(sql)) {
      print(resultSet);
    }
}

输出结果 (成功)：

10, Sales_addStr
20, Marketing_addStr
30, Accounts_addStr

然而，如果启用校验方法 validateSql(...)，则会遇到与尝试一相同的校验错误。这是因为校验器 SqlValidator 默认使用 SqlStdOperatorTable.instance()，它不认识我们通过 Schema 注册的非标准函数。

4. 问题根源与解决思路

两种尝试揭示了 Calcite 中 UDF 处理的“生命周期”差异：

• SqlFunction + SqlOperatorTable： 作用于解析/校验阶段，定义了“函数签名”。
• Schema.add(...)： 作用于执行阶段，绑定了“函数实现”。

要使 UDF 在 Calcite 中全程可用（即可校验、可执行），需要将两种方式结合：

1. 声明阶段： 确保校验器 (SqlValidator) 能识别函数签名。这可以通过扩展 SqlOperatorTable（如尝试一），或者在创建校验器时，将 Schema 中的函数也注入到校验上下文中来实现。
2. 执行阶段： 确保执行引擎能调用到函数的具体实现。这通过将 Java 方法注册到 Schema（如尝试二）来完成。

解决此问题的关键是统一校验阶段和执行阶段对 UDF 的认知。一种方案是创建自定义的 SqlValidator 或 CalciteCatalogReader，使其在查找函数时，不仅能查询标准的 SqlStdOperatorTable，也能查询当前 Schema 中注册的用户函数。这需要对 Calcite 的后端 & 架构有更深的理解，通过探索其扩展机制来桥接声明与执行两个环节，从而实现完整的、支持校验的自定义 UDF 功能。