openHalo兼容mysql原理和实践

NickYoung

发布于 2026-07-09 16:30:56

最近老杨体验了多个基于PG的开源数据库，对于有鲜明特点的开源产品总会抽时间上手把玩下。

本期我们来看看openHalo，在同一个实例中PG和MySQL一起玩，挺有意思的。

兼容性简介

从官方介绍[1]来看，openHalo基于PG，同时能够理解MySQL的SQL方言，并支持相同的通信协议，因此最初为MySQL编写的应用程序只需进行少量代码更改即可与openHalo兼容。

PG和MySQL协议各自监听一个port，可以使用对应客户端访问对应端口进行操作。

MySQL兼容实现原理：大概看了下代码，引入MySQL协议，在parser里通过makeFuncCall或者sql转换等方式将MySQL方言和sql转换为PG协议再去执行。

具体可以参考以下目录中的源代码： src/backend/parser/mysql src/backend/adapter/mysql

兼容性验证

使用mysql协议登录后，进行简单验证。

show databases; use database; show tables; show create table; ... 这些命令的返回都和mysql一致。

但show processlist;输出的是固定的结果，不能查看实时会话。

mysql> show processlist;
+----+------+------+----+---------+------+-------+------+
| Id | User | Host | db | Command | Time | State | Info |
+----+------+------+----+---------+------+-------+------+
|  1 |      |      |    |         |    0 |       |      |
+----+------+------+----+---------+------+-------+------+
1 row in set (0.00 sec)

mysql> showprocesslist;
+----+------+------+----+---------+------+-------+------+
| Id | User | Host | db | Command | Time | State | Info |
+----+------+------+----+---------+------+-------+------+
|  1 |      |      |    |         |    0 |       |      |
+----+------+------+----+---------+------+-------+------+
1 row in set (0.00 sec)

mysql>

这对于习惯使用processlist的朋友来说可能会有些不方便。

我举个简单例子，目前openHalo的架构下，我们建一个表，这个表既可以通过mysql协议访问，也可以通过PG协议访问。如何去排查并发访问锁冲突问题呢？

一般MySQL中大家可能习惯了使用show processlist，但在openHalo里查不了？

非也，查询pg_stat_activity即可。你没看错，mysql工具登录可以进行一些PG的视图查询。基于PG引进了MySQL协议，根据目前兼容性原理是可以查询一些PG的系统视图等对象的。

mysql> show processlist;
+----+------+------+----+---------+------+-------+------+
| Id | User | Host | db | Command | Time | State | Info |
+----+------+------+----+---------+------+-------+------+
|  1 |      |      |    |         |    0 |       |      |
+----+------+------+----+---------+------+-------+------+
1 row in set (0.00 sec)

mysql> select * from pg_stat_activity where pid=pg_backend_pid();
+-------+-----------+--------+------------+----------+----------+------------------+-------------+-----------------+-------------+-------------------------------+-------------------------------+-------------------------------+-------------------------------+-----------------+------------+--------+-------------+--------------+----------+-----------------------------------------------------------+----------------+
| datid | datname   | pid    | leader_pid | usesysid | usename  | application_name | client_addr | client_hostname | client_port | backend_start                 | xact_start                    | query_start                   | state_change                  | wait_event_type | wait_event | state  | backend_xid | backend_xmin | query_id | query                                                     | backend_type   |
+-------+-----------+--------+------------+----------+----------+------------------+-------------+-----------------+-------------+-------------------------------+-------------------------------+-------------------------------+-------------------------------+-----------------+------------+--------+-------------+--------------+----------+-----------------------------------------------------------+----------------+
| 13889 | halo0root | 483422 |       NULL | 10       | openhalo |                  | 127.0.0.1   | NULL            |       38042 | 2025-11-06 11:33:14.312263+08 | 2025-11-06 16:50:15.244951+08 | 2025-11-06 16:50:15.244951+08 | 2025-11-06 16:50:15.244954+08 | NULL            | NULL       | active | NULL        | 750          |     NULL | select * from pg_stat_activity where pid=pg_backend_pid() | client backend |
+-------+-----------+--------+------------+----------+----------+------------------+-------------+-----------------+-------------+-------------------------------+-------------------------------+-------------------------------+-------------------------------+-----------------+------------+--------+-------------+--------------+----------+-----------------------------------------------------------+----------------+
1rowinset (0.00 sec)

mysql>

兼容show processlist

现在的实现是这样的： Id赋值1，time赋值0，其他都为NULL。

            | SHOW ShowFullProcesslistOpt PROCESSLIST
                {
                    /* SHOW [FULL] PROCESSLIST */
                    /* "select NULL::int as Id, ' '::varchar(256) as User, ' '::varchar(256) as Host, 
                     * ' '::varchar(256) as db, ' '::varchar(256) as Command, ' '::bigint as Time, 
                     * ' '::varchar(256) as State, ' '::varchar(256) as Info"); 
                     */
                                        SelectStmt *selectStmt;
                                        selectStmt = makeNode(SelectStmt);
                    selectStmt->targetList = NIL;
                    selectStmt->targetList = lappend(selectStmt->targetList, createResTargetWithInt("Id", 1));
                    selectStmt->targetList = lappend(selectStmt->targetList, createResTargetWithString("User", " "));
                    selectStmt->targetList = lappend(selectStmt->targetList, createResTargetWithString("Host", " "));
                    selectStmt->targetList = lappend(selectStmt->targetList, createResTargetWithString("db", " "));
                    selectStmt->targetList = lappend(selectStmt->targetList, createResTargetWithString("Command", " "));
                    selectStmt->targetList = lappend(selectStmt->targetList, createResTargetWithInt("Time", 0));
                    selectStmt->targetList = lappend(selectStmt->targetList, createResTargetWithString("State", " "));
                    selectStmt->targetList = lappend(selectStmt->targetList, createResTargetWithString("Info", " "));
                                        $$ = (Node *)selectStmt;
                }

了解了兼容性原理后，这个实现起来也很简单，创建函数mysql.show_processlist()用来查询pg_stat_activity匹配processlist输出的字段。不过db字段没法匹配，都会显示PG协议下的datname，另外这里的id就是进程id了。

aux_mysql--1.3--1.4.sql:

CREATE OR REPLACE FUNCTION mysql.show_processlist()
RETURNS SETOF record
LANGUAGE plpgsql
STABLE
AS $function$
DECLARE
    result_record RECORD;
BEGIN
    FOR result_record IN
        SELECT
            pid::integer as "Id",                  
            usename::text as "User",                
            CASE 
                WHEN client_addr IS NULL THEN 'local'::text
                ELSE pg_catalog.concat(client_addr::text, ':', client_port::text)
            END as "Host",
            datname::text as "db",                  
            CASE 
                WHEN query ILIKE 'SELECT%' THEN 'SELECT'::text
                WHEN query ILIKE 'INSERT%' THEN 'INSERT'::text
                WHEN query ILIKE 'UPDATE%' THEN 'UPDATE'::text
                WHEN query ILIKE 'DELETE%' THEN 'DELETE'::text
                WHEN query ILIKE 'CREATE%' THEN 'CREATE'::text
                WHEN query ILIKE 'ALTER%' THEN 'ALTER'::text
                WHEN query ILIKE 'DROP%' THEN 'DROP'::text
                WHEN query ILIKE 'BEGIN%' THEN 'BEGIN'::text
                WHEN query ILIKE 'COMMIT%' THEN 'COMMIT'::text
                WHEN query ILIKE 'ROLLBACK%' THEN 'ROLLBACK'::text
                WHEN query ILIKE 'TRUNCATE%' THEN 'TRUNCATE'::text
                ELSE 'OTHER'::text
            END AS "Command",
            (now() - query_start)::text as "Time",  
            CASE 
                WHEN state = 'active' AND wait_event_type IS NULL THEN 'executing'
                WHEN state = 'active' AND wait_event_type IS NOT NULL THEN 
                   pg_catalog.concat('waiting for ', wait_event_type, '.', wait_event)
                WHEN state = 'idle' THEN 'sleep'
                WHEN state = 'idle' AND wait_event_type IS NOT NULL THEN 
                   pg_catalog.concat('waiting for ', wait_event_type, '.', wait_event)
                WHEN wait_event_type IS NOT NULL THEN 
                   pg_catalog.concat('waiting for ', wait_event_type, '.', wait_event)
            ELSE state
        END::text AS "State",                    
            LEFT(COALESCE(query, ''), 300)::text as "Info"
        FROM pg_catalog.pg_stat_activity
        WHERE datname IS NOT NULL
    LOOP
        RETURN NEXT result_record;
    END LOOP;
    
    RETURN;
END;
$function$;

mys_gram.y:

| SHOW ShowFullProcesslistOpt PROCESSLIST
                {
                    /* SHOW [FULL] PROCESSLIST */
                    /* select mysql.show_processlist(); */
                    SelectStmt *selectStmt;
                    RangeFunction *rangeFunction;
                    FuncCall *funcCall;
                    List *funcAliasClause;
                    List *tableFuncElementList;

                    selectStmt = makeNode(SelectStmt);
                    selectStmt->targetList = NIL;
                    selectStmt->targetList = lappend(selectStmt->targetList, createResTargetWithStart());  
    
                    rangeFunction = makeNode(RangeFunction);
                    rangeFunction->lateral = false;
                    rangeFunction->ordinality = false;
                    rangeFunction->is_rowsfrom = false;
    
                    /* function call: mysql.show_processlist() */
                    funcCall = makeFuncCall(
                        list_make2(makeString(pstrdup("mysql")),
                    makeString(pstrdup("show_processlist"))),
                    NIL,  
                    COERCE_EXPLICIT_CALL,
                    -1
                    );
    
                    rangeFunction->functions = list_make1(list_make2((Node *)funcCall, NIL));
    
                    tableFuncElementList = NIL;
                    tableFuncElementList = lappend(tableFuncElementList, (Node *)createColumnDef("Id", "int4", 0));
                    tableFuncElementList = lappend(tableFuncElementList, (Node *)createColumnDef("User", "text", 0));
                    tableFuncElementList = lappend(tableFuncElementList, (Node *)createColumnDef("Host", "text", 0));
                    tableFuncElementList = lappend(tableFuncElementList, (Node *)createColumnDef("db", "text", 0));
                    tableFuncElementList = lappend(tableFuncElementList, (Node *)createColumnDef("Command", "text", 0));
                    tableFuncElementList = lappend(tableFuncElementList, (Node *)createColumnDef("Time", "text", 0));
                    tableFuncElementList = lappend(tableFuncElementList, (Node *)createColumnDef("State", "text", 0));
                    tableFuncElementList = lappend(tableFuncElementList, (Node *)createColumnDef("Info", "text", 0));
    
                    funcAliasClause = list_make2(NULL, tableFuncElementList);
                    rangeFunction->alias = linitial(funcAliasClause);
                    rangeFunction->coldeflist = lsecond(funcAliasClause);
    
                    selectStmt->fromClause = list_make1((Node *)rangeFunction);
    
                    $$ = (Node *)selectStmt;
                }

重新编译安装后，进行测试。

PG协议会话执行truncate table mys_sys.test长事务未提交，MySQL协议会话在mys_sys中select * from test;

show processlist可以看到所有会话信息，我们通过State和Info就能得知select被truncate阻塞。

  
mysql> showprocesslist;
+-------+------+--------------------+-----------+----------+-----------------+-------------------------------+---------------------------------+
| Id    | User | Host               | db        | Command  | Time            | State                         | Info                            |
+-------+------+--------------------+-----------+----------+-----------------+-------------------------------+---------------------------------+
| 57248 | halo | 127.0.0.1/32:44958 | halo0root | SELECT   | 00:00:29.74024  | sleep                         | select * from pg_stat_activity; |
| 55582 | halo | 127.0.0.1/32:59904 | halo0root | OTHER    | 00:00:00.374257 | executing                     | show processlist               |
| 56831 | halo | 127.0.0.1/32:34544 | halo0root | SELECT   | 00:01:10.07766  | waiting for Lock.relation     | select * from test              |
| 56546 | halo | 127.0.0.1/32:60472 | halo0root | TRUNCATE | 00:02:28.987154 | waiting for Client.ClientRead | truncate TABLE mys_sys.test ;   |
+-------+------+--------------------+-----------+----------+-----------------+-------------------------------+---------------------------------+
4 rows in set (0.00 sec)

mysql>