开发环境准备

本次连接器开发基于以下环境：

• WSL (Ubuntu 18.04.5) + WSLg，配合支持 GUI 的 IDE 进行调试
• Maven 3.6.3
• OpenJDK 1.8.0_312
• Docker 20.10.14（Docker Desktop 4.8.2 集成 WSL）

设计思路

MariaDB 作为 MySQL 的分支，完全兼容 JDBC 协议。openLooKeng 已为 JDBC 数据源封装了统一的 BaseJdbcXXX 抽象层，因此可以借鉴 MySQL 连接器的实现方式，通过扩展这些基础 API 来完成 MariaDB 连接器的开发。

功能验证

部署完成后启动 openLooKeng 服务：

$ ./launcher start
Started as 6874

通过 CLI 工具连接并查询：

$ java -jar hetu-cli-1.7.0-SNAPSHOT-executable.jar --server localhost:8080 --catalog maria --schema test

lk:test> select * from user;
 id |    name
----+-------------
  1 | tom
  2 | link
  3 | maria
  4 | xiaohuangya
  6 | xiaohuangya
  9 | hyy
(6 rows)

项目初始化

从 Gitee 拉取 openLooKeng 1.7.0 源码，先执行依赖安装：

mvn clean install -DskipTests

创建模块

新建 hetu-mariadb 模块，parent 指向 presto-root，packaging 设为 hetu-plugin，确保打包后自动归入 heto-core 的 plugin 目录。

核心依赖配置

参考 MySQL 连接器的依赖结构，关键区别在于引入 MariaDB 官方驱动：

<dependency>
    <groupId>org.mariadb.jdbc</groupId>
    <artifactId>mariadb-java-client</artifactId>
    <version>2.7.3</version>
</dependency>

同时依赖 presto-base-jdbc 等基础组件，测试相关依赖包括 testing-mysql-server、presto-testing-docker 等。

插件注册

在 hetu-server 的打包配置中注册该连接器：

<artifactSet to="plugin/mariadb">
    <artifact id="${project.groupId}:hetu-mariadb:zip:${project.version}">
        <unpack/>
    </artifact>
</artifactSet>

核心实现详解

外部函数下推

继承 ApplyRemoteFunctionPushDown 实现下推逻辑：

public class MariaApplyRemoteFunctionPushDown
        extends ApplyRemoteFunctionPushDown
{
    public MariaApplyRemoteFunctionPushDown(BaseJdbcConfig baseJdbcConfig, String connectorName)
    {
        super(baseJdbcConfig, connectorName);
    }
}

外部函数注册

通过 JdbcExternalFunctionHub 绑定数学、字符串、日期三类函数：

public class MariaExternalFunctionHub
        extends JdbcExternalFunctionHub
{
    private final BaseJdbcConfig jdbcConfig;

    @Inject
    public MariaExternalFunctionHub(BaseJdbcConfig jdbcConfig)
    {
        this.jdbcConfig = requireNonNull(jdbcConfig, "jdbcConfig is null");
    }

    @Override
    public Optional<CatalogSchemaName> getExternalFunctionCatalogSchemaName()
    {
        return jdbcConfig.getConnectorRegistryFunctionNamespace();
    }

    @Override
    public Set<ExternalFunctionInfo> getExternalFunctions()
    {
        return ImmutableSet.<ExternalFunctionInfo>builder()
                .addAll(MariaExternalMathFunctions.getFunctionsInfo())
                .addAll(MariaExternalStringFunctions.getFunctionsInfo())
                .addAll(MariaExternalDateTimeFunctions.getFunctionsInfo())
                .build();
    }
}

日期函数示例：

public final class MariaExternalDateTimeFunctions
{
    private static final ExternalFunctionInfo MARIA_MONTH_NAME_FUNCTION_INFO = ExternalFunctionInfo.builder()
            .functionName("monthname")
            .inputArgs(StandardTypes.DATE)
            .returnType(StandardTypes.VARCHAR)
            .deterministic(true)
            .calledOnNullInput(false)
            .description("return name of the month for the input date")
            .build();

    private static final ExternalFunctionInfo MARIA_HOUR_FUNCTION_INFO = ExternalFunctionInfo.builder()
            .functionName("hour")
            .inputArgs(StandardTypes.TIME)
            .returnType(StandardTypes.INTEGER)
            .deterministic(true)
            .calledOnNullInput(false)
            .description("return hour value for the input time")
            .build();

    private static final ExternalFunctionInfo MARIA_TIMESTAMP_FUNCTION_INFO = ExternalFunctionInfo.builder()
            .functionName("timestamp")
            .inputArgs(StandardTypes.TIMESTAMP, StandardTypes.TIME)
            .returnType(StandardTypes.TIMESTAMP)
            .deterministic(true)
            .calledOnNullInput(false)
            .description("adds time expr2 to expr1 and return the result as a timestamp value")
            .build();

    private static final ExternalFunctionInfo MARIA_DATE_FORMAT_FUNCTION_INFO = ExternalFunctionInfo.builder()
            .functionName("date_format")
            .inputArgs(StandardTypes.TIMESTAMP, StandardTypes.VARCHAR)
            .returnType(StandardTypes.VARCHAR)
            .deterministic(true)
            .calledOnNullInput(false)
            .description("format the date value according to the format string")
            .build();

    public static Set<ExternalFunctionInfo> getFunctionsInfo()
    {
        return ImmutableSet.<ExternalFunctionInfo>builder()
                .add(MARIA_MONTH_NAME_FUNCTION_INFO)
                .add(MARIA_HOUR_FUNCTION_INFO)
                .add(MARIA_TIMESTAMP_FUNCTION_INFO)
                .add(MARIA_DATE_FORMAT_FUNCTION_INFO)
                .build();
    }

    private MariaExternalDateTimeFunctions() {}
}

数学函数示例：

public final class MariaExternalMathFunctions
{
    private static final ExternalFunctionInfo MARIA_TRUNCATE_FUNCTION_INFO = ExternalFunctionInfo.builder()
            .functionName("truncate")
            .inputArgs(StandardTypes.DOUBLE, StandardTypes.INTEGER)
            .returnType(StandardTypes.DOUBLE)
            .deterministic(true)
            .calledOnNullInput(false)
            .description("return the value x that is reserved to y decimal places")
            .build();

    private static final ExternalFunctionInfo MARIA_ABS_FUNCTION_INFO = ExternalFunctionInfo.builder()
            .functionName("abs")
            .inputArgs(StandardTypes.INTEGER)
            .returnType(StandardTypes.INTEGER)
            .deterministic(true)
            .calledOnNullInput(false)
            .description("return the absolute value of x")
            .build();

    private static final ExternalFunctionInfo MARIA_BIG_INT_ABS_FUNCTION_INFO = ExternalFunctionInfo.builder()
            .functionName("abs")
            .inputArgs(StandardTypes.BIGINT)
            .returnType(StandardTypes.BIGINT)
            .deterministic(true)
            .calledOnNullInput(false)
            .description("return the absolute value of x")
            .build();

    public static Set<ExternalFunctionInfo> getFunctionsInfo()
    {
        return ImmutableSet.<ExternalFunctionInfo>builder()
                .add(MARIA_TRUNCATE_FUNCTION_INFO)
                .add(MARIA_ABS_FUNCTION_INFO)
                .add(MARIA_BIG_INT_ABS_FUNCTION_INFO)
                .build();
    }

    private MariaExternalMathFunctions() {}
}

字符串函数示例：

public final class MariaExternalStringFunctions
{
    private static final ExternalFunctionInfo MARIA_FORMAT_FUNCTION_INFO = ExternalFunctionInfo.builder()
            .functionName("format")
            .inputArgs(StandardTypes.DOUBLE, StandardTypes.INTEGER)
            .returnType(StandardTypes.VARCHAR)
            .deterministic(true)
            .calledOnNullInput(false)
            .description("format the number 'num' to a format like'#,###,###.##', rounded to 'lo' decimal places, and returns the result as a string")
            .build();

    private static final ExternalFunctionInfo MARIA_FORMAT_DECIMAL_FUNCTION_INFO = ExternalFunctionInfo.builder()
            .functionName("format")
            .inputArgs(StandardTypes.DECIMAL, StandardTypes.INTEGER)
            .returnType(StandardTypes.VARCHAR)
            .deterministic(true)
            .calledOnNullInput(false)
            .description("format the number 'num' to a format like'#,###,###.##', rounded to 'lo' decimal places, and returns the result as a string")
            .build();

    private static final ExternalFunctionInfo MARIA_LOWER_FUNCTION_INFO = ExternalFunctionInfo.builder()
            .functionName("lower")
            .inputArgs(StandardTypes.VARCHAR)
            .returnType(StandardTypes.VARCHAR)
            .deterministic(true)
            .calledOnNullInput(false)
            .description("returns the string str with all characters changed to lowercase")
            .build();

    public static Set<ExternalFunctionInfo> getFunctionsInfo()
    {
        return ImmutableSet.<ExternalFunctionInfo>builder()
                .add(MARIA_FORMAT_FUNCTION_INFO)
                .add(MARIA_FORMAT_DECIMAL_FUNCTION_INFO)
                .add(MARIA_LOWER_FUNCTION_INFO)
                .build();
    }

    private MariaExternalStringFunctions() {}
}

插件入口类

public class MariaPlugin extends JdbcPlugin
{
    public MariaPlugin()
    {
        super("maria", new MariaClientModule());
    }

    @Override
    public Optional<ConnectorWithProperties> getConnectorWithProperties()
    {
        ConnectorConfig connectorConfig = MariaPlugin.class.getAnnotation(ConnectorConfig.class);
        Optional<ConnectorWithProperties> connectorWithProperties = ConnectorUtil.assembleConnectorProperties(
                connectorConfig, 
                Arrays.asList(BaseJdbcConfig.class.getDeclaredMethods()));
        ConnectorUtil.addConnUrlProperty(connectorWithProperties, "jdbc:mariadb://host:port");
        return connectorWithProperties;
    }
}

客户端模块绑定

MariaClientModule 负责绑定核心组件，并处理 URL 校验逻辑。MariaDB 驱动的 URL 解析由 URLParser 完成，与 MySQL 驱动直接在 driver 类中处理不同：

public class MariaClientModule extends AbstractConfigurationAwareModule
{
    @Override
    protected void setup(Binder binder)
    {
        binder.bind(JdbcClient.class).to(MariaClient.class).in(Scopes.SINGLETON);
        binder.bind(ExternalFunctionHub.class).to(MariaExternalFunctionHub.class).in(Scopes.SINGLETON);
        ensureCatalogIsEmpty(buildConfigObject(BaseJdbcConfig.class).getConnectionUrl());
        configBinder(binder).bindConfig(MariaConfig.class);
    }

    private static void ensureCatalogIsEmpty(String connectionUrl)
    {
        try {
            UrlParser urlParser = UrlParser.parse(connectionUrl);
            checkArgument(urlParser != null, "Invalid JDBC URL for MariaDB connector");
            checkArgument(urlParser.getDatabase() == null, 
                    "Database (catalog) must not be specified in JDBC URL for MariaDB connector");
        }
        catch (SQLException e) {
            throw new RuntimeException(e);
        }
    }

    @Provides
    @Singleton
    public static ConnectionFactory createConnectionFactory(BaseJdbcConfig config, MariaConfig mariaConfig)
            throws SQLException
    {
        Properties connectionProperties = basicConnectionProperties(config);
        connectionProperties.setProperty("useInformationSchema", "true");
        connectionProperties.setProperty("nullCatalogMeansCurrent", "false");
        connectionProperties.setProperty("useUnicode", "true");
        connectionProperties.setProperty("characterEncoding", "utf8");
        connectionProperties.setProperty("tinyInt1isBit", "false");

        if (mariaConfig.isAutoReconnect()) {
            connectionProperties.setProperty("autoReconnect", String.valueOf(mariaConfig.isAutoReconnect()));
            connectionProperties.setProperty("maxReconnects", String.valueOf(mariaConfig.getMaxReconnects()));
        }
        if (mariaConfig.getConnectionTimeout() != null) {
            connectionProperties.setProperty("connectTimeout", 
                    String.valueOf(miaConfig.getConnectionTimeout().toMillis()));
        }

        return new DriverConnectionFactory(
                new Driver(),
                config.getConnectionUrl(),
                Optional.ofNullable(config.getUserCredentialName()),
                Optional.ofNullable(config.getPasswordCredentialName()),
                connectionProperties);
    }
}

客户端核心实现

MariaClient 基于 BaseJdbcClient 针对 MariaDB 特性进行适配。

Schema 列表获取：MariaDB 中需获取 catalog 而非 schema：

@Override
protected Collection<String> listSchemas(Connection connection)
{
    try (ResultSet resultSet = connection.getMetaData().getCatalogs()) {
        ImmutableSet.Builder<String> schemaNames = ImmutableSet.builder();
        while (resultSet.next()) {
            String schemaName = resultSet.getString("TABLE_CAT");
            if (!schemaName.equalsIgnoreCase("information_schema") 
                    && !schemaName.equalsIgnoreCase("mysql")) {
                schemaNames.add(schemaName);
            }
        }
        return schemaNames.build();
    }
    catch (SQLException e) {
        throw new RuntimeException(e);
    }
}

预编译语句封装：

@Override
public PreparedStatement getPreparedStatement(Connection connection, String sql) throws SQLException
{
    PreparedStatement statement = connection.prepareStatement(sql);
    if (statement.isWrapperFor(MariaDbStatement.class)) {
        statement.unwrap(MariaDbStatement.class);
    }
    return statement;
}

异常处理：MariaDB 驱动未提供 ErrorCode 与 SqlState 的映射，需通过异常 cause 判断：

@Override
public void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
{
    try {
        createTable(session, tableMetadata, tableMetadata.getTable().getTableName());
    }
    catch (PrestoException e) {
        if (e.getCause() instanceof SQLSyntaxErrorException) {
            throw new PrestoException(ALREADY_EXISTS, format("Table already exists", e));
        }
    }
    catch (SQLException e) {
        throw new PrestoException(JDBC_ERROR, e);
    }
}

@Override
public void renameColumn(JdbcIdentity identity, JdbcTableHandle handle, 
        JdbcColumnHandle jdbcColumn, String inputNewColumnName)
{
    String newColumnName = inputNewColumnName;
    try (Connection connection = connectionFactory.openConnection(identity)) {
        DatabaseMetaData metadata = connection.getMetaData();
        if (metadata.storesUpperCaseIdentifiers()) {
            newColumnName = newColumnName.toUpperCase(ENGLISH);
        }
        String sql = format("ALTER TABLE %s RENAME COLUMN %s TO %s",
                quoted(handle.getCatalogName(), handle.getSchemaName(), handle.getTableName()),
                quoted(jdbcColumn.getColumnName()),
                quoted(newColumnName));
        execute(connection, sql);
    }
    catch (PrestoException e) {
        if (e.getCause() instanceof SQLSyntaxErrorException) {
            throw new PrestoException(NOT_SUPPORTED, 
                    format("Rename column not supported for the MariaDB server version", e));
        }
    }
    catch (SQLException e) {
        throw new PrestoException(JDBC_ERROR, e);
    }
}

查询生成器：

@Override
public Optional<QueryGenerator<JdbcQueryGeneratorResult, JdbcConverterContext>> getQueryGenerator(
        DeterminismEvaluator determinismEvaluator,
        RowExpressionService rowExpressionService,
        FunctionMetadataManager functionManager,
        StandardFunctionResolution functionResolution)
{
    JdbcPushDownModule pushDownModule = pushDownModule == DEFAULT ? BASE_PUSHDOWN : pushDownModule;
    JdbcPushDownParameter pushDownParameter = new JdbcPushDownParameter(
            getIdentifierQuote(),
            this.caseInsensitiveNameMatching,
            pushDownModule,
            functionResolution);
    return Optional.of(new MariaQueryGenerator(
            determinismEvaluator,
            rowExpressionService,
            functionManager,
            functionResolution,
            pushDownParameter,
            config));
}

下推策略枚举：

public enum JdbcPushDownModule
{
    DEFAULT,
    FULL_PUSHDOWN,
    BASE_PUSHDOWN;

    private static final Set<Class<? extends PlanNode>> BASE_PUSH_DOWN_NODE = ImmutableSet.of(
            FilterNode.class,
            AggregationNode.class,
            LimitNode.class,
            TopNNode.class,
            ProjectNode.class,
            TableScanNode.class);

    public boolean isAvailable(PlanNode node)
    {
        switch (this) {
            case FULL_PUSHDOWN:
                return true;
            case BASE_PUSHDOWN:
                return BASE_PUSH_DOWN_NODE.contains(node.getClass());
            default:
                throw new PrestoException(NOT_SUPPORTED, "Unsupported push down module");
        }
    }
}

测试体系

插件基础测试

public class TestMariaPlugin
{
    @Test
    public void testCreateConnector()
    {
        Plugin plugin = new MariaPlugin();
        ConnectorFactory factory = getOnlyElement(plugin.getConnectorFactories());
        factory.create("test", ImmutableMap.of("connection-url", "jdbc:mariadb://test"), 
                new TestingConnectorContext());
    }
}

配置映射测试

public class TestMariaConfig
{
    @Test
    public void testDefaults()
    {
        assertRecordedDefaults(recordDefaults(MariaConfig.class)
                .setAutoReconnect(true)
                .setMaxReconnects(3)
                .setConnectionTimeout(new Duration(10, TimeUnit.SECONDS)));
    }

    @Test
    public void testExplicitPropertyMappings()
    {
        Map<String, String> properties = new ImmutableMap.Builder<String, String>()
                .put("mysql.auto-reconnect", "false")
                .put("mysql.max-reconnects", "4")
                .put("mysql.connection-timeout", "4s")
                .build();

        MariaConfig expected = new MariaConfig()
                .setAutoReconnect(false)
                .setMaxReconnects(4)
                .setConnectionTimeout(new Duration(4, TimeUnit.SECONDS));

        assertFullMapping(properties, expected);
    }
}

查询执行器

基于 airlift 的 TestingMySqlServer 自动创建测试数据库实例，无需手动配置外部环境：

public static final TestingMySqlServer createTestingMySqlServer(
        String userName, String password, String schemaName, String dataBase) throws Exception
{
    TestingMySqlServer tempServer = null;
    final int createRetry = 3;

    for (int i = 0; i < createRetry; i++) {
        try {
            if (dataBase == null) {
                tempServer = new TestingMySqlServer(userName, password, schemaName);
            } else {
                tempServer = new TestingMySqlServer(userName, password, schemaName, dataBase);
            }
        }
        catch (Exception e) {
            if (i == (createRetry - 1)) {
                throw e;
            }
            continue;
        }
        break;
    }
    return tempServer;
}

Docker 容器化测试

public class DockerizedMariaServer implements Closeable
{
    private static final int MYSQL_PORT = 3306;
    private static final String MYSQL_ROOT_USER = "root";
    private static final String MYSQL_ROOT_PASSWORD = "mysqlrootpassword";
    private static final String MYSQL_USER = "testuser";
    private static final String MYSQL_PASSWORD = "testpassword";

    private final DockerContainer dockerContainer;

    public DockerizedMariaServer()
    {
        try {
            this.dockerContainer = new DockerContainer(
                    "mariadb",
                    ImmutableList.of(MYSQL_PORT),
                    ImmutableMap.of(
                            "MYSQL_ROOT_PASSWORD", MYSQL_ROOT_PASSWORD,
                            "MYSQL_USER", MYSQL_USER,
                            "MYSQL_PASSWORD", MYSQL_PASSWORD,
                            "MYSQL_DATABASE", "tpch"),
                    DockerizedMariaServer::healthCheck);
        }
        catch (Exception e) {
            System.out.println("## Docker environment not properly set up. Skip test. ##");
            throw new SkipException("Docker environment not initialized for tests");
        }
    }
}

分布式查询测试

修复 getColumns 方法中的空指针问题，增加对 ResultSetMetaData 的 null 校验：

@Override
public Map<String, ColumnHandle> getColumns(
        ConnectorSession session, String sql, Map<String, Type> types)
{
    try (Connection connection = connectionFactory.openConnection(JdbcIdentity.from(session));
         PreparedStatement statement = connection.prepareStatement(sql)) {
        ResultSetMetaData metaData = statement.getMetaData();
        ImmutableMap.Builder<String, ColumnHandle> columnBuilder = new ImmutableMap.Builder<>();
        if (metaData != null) {
            // ... 列信息处理逻辑
        }
        // ...
    }
}

集成冒烟测试

@Test
public void integerColumn() throws Exception
{
    Session session = testSessionBuilder()
            .setCatalog("maria")
            .setSchema("test_database")
            .build();

    assertUpdate("CREATE TABLE maria.test_database.testInteger(" +
            "a tinyint, b smallint, c integer, d bigint)");
    assertUpdate("INSERT INTO maria.test_database.testInteger VALUES " +
            "(tinyint '-2', smallint '1', 5, 158)", 1);
    assertQuery("SELECT * FROM maria.test_database.testInteger",
            "SELECT -2 a, 1 b, 5 c, 158 d");
    assertUpdate("DROP TABLE maria.test_database.testInteger");
}

常见问题

编译失败

首次编译需跳过测试并安装依赖：

mvn clean install -DskipTests

权限不足时使用 sudo 执行。

mysqld root 身份启动受限

错误信息：Please read "Security" section of the manual to find out how to run mysqld as root!

解决方案：

• 以普通用户身份启动 IDE 运行测试
• 或修改 io.airlift.testing.mysql.EmbeddedMySql 源码，在 startMysqld 方法中追加参数 --user=root

测试环境说明

单元测试运行无需本地安装 MySQL 或 MariaDB，TestingMySqlServer 会自动创建 mysqld 进程。DockerizedMariaServer 则依赖本地 Docker 服务，测试时会自动拉取镜像并启动容器。