opengauss-serveer多数据库网络协议扩展支持

**Is this a BUG REPORT or FEATURE REQUEST?**:

> - feature

## 1.特性概述

*同MySQL相比，opengauss拥有更加优秀的性能及扩展性，但国内数据库市场早期以Oracle、MySQL为主，经过多年高速增长，已经形成了庞大的用户基数；通过在opengauss server 层实现多网络协议的扩展框架后，可以让opengauss原生兼容其它数据库社区的生态，对于MySQL用户来说，可以在不改变操作习惯的前提下，平滑过渡到opengauss生态体系，它既可以是一个性能加强版本的MySQL，也是一个完整的opengauss。*

## 2.目标

*通过在opengauss内核层抽象一层网络协议交互接口，让opengauss具备以插件的方式动态支持多数据库网络协议的能力。*

## 3.特性需求列表

| 需求编号 | 需求名称 | 特性描述 | 备注  |
| --- | --- | --- | --- |
| 1   | 协议扩展接口抽象 | 在opengauss内核提供协议扩展接口 | SQL引擎 SIG组评审 |
| 2   | opengauss协议层包装 | 基于协议扩展接口轻量包装既有opengauss协议相关逻辑 | SQL引擎 SIG组评审  |
| 3   | MySQL协议插件实现 | 插件化方式提供MySQL协议扩展接口实现 | Plugin SIG组评审 |

## 4.总体方案

*方案整体架构图，以opengauss兼容MySQL网络协议为例：*

*<img src="https://foruda.gitee.com/images/1662086734464648109/669b139b_5692953.png" width="310" height="429" class="jop-noMdConv">*

*当首次创建/使用B兼容类型的数据库后，默认的MySQL 3306监听端口将会被自动拉起，MySQL的网络协议实现也会被绑定到3306 端口所对应的fd中；当用户通过MySQL driver 连接opengauss的3306端口时，opengauss的网络协议处理层，选择监听端口fd所对应的ProtocolExtensionConfig实现来响应这次网络请求。*

*<img src="https://foruda.gitee.com/images/1662086932321213709/ff03d9df_5692953.png" width="668" height="285" class="jop-noMdConv">*

## 5.设计思路

通过在port上绑定ProtocolExtensionConfig接口，来区分不通数据库网络协议在opengauss的处理流程；ProtocolExtensionConfig接口的默认实现方式为opengauss的网络协议，当首次创建/使用B兼容类型数据库时，*MySQL相关的端口和ProtocolExtensionConfig实现将自动进行创建和绑定，这样用户可以通过MySQL的客户端直连opengauss的MySQL监听端口，可以原生兼容MySQL Client/Server间的网络交互协议。*

## 6. 详细实现(从用户入口的模块级别或进程级别消息序列图)

*<img src="https://foruda.gitee.com/images/1662087296237965660/4a6f01d1_5692953.png" width="1023" height="543" class="jop-noMdConv">*

*ProtocolExtensionConfig定义了网络协议交互的action，包含如下方法：*

```C++
typedef struct ProtocolExtensionConfig {
    bool    server_handshake_first;
    int     (*fn_accept)(pgsocket server_fd, struct Port *port);
    void    (*fn_close)(pgsocket server_fd);
    void    (*fn_init)(void);
    int     (*fn_start)(struct Port *port);
    void    (*fn_authenticate)(struct Port *port);
    void    (*fn_send_message)(ErrorData *edata);
    void    (*fn_send_cancel_key)(int32 pid, int32 key);
    void    (*fn_comm_reset)(void);
    void    (*fn_send_ready_for_query)(CommandDest dest);
    int	    (*fn_read_command)(StringInfo inBuf);
    void    (*fn_end_command)(const char *tag, CommandDest dest);
    DestReceiver*   (*fn_printtup_create_DR)(CommandDest dest);
    void    (*fn_set_DR_params)(DestReceiver* self, List* target_list);
    int     (*fn_process_command)(StringInfo inBuf);
    void    (*fn_report_param_status)(const char *name, char *val);
} ProtocolExtensionConfig;
```

其中，默认的opengauss网络协议的实现方式为：

```C++
ProtocolExtensionConfig default_protocol_config = {
    false,
    StreamConnection,
    StreamClose,
    pq_init,
    StartupPacketInitialize,
    ClientAuthentication,
    send_message_to_frontend,
    libpq_send_cancel_key,
    pq_comm_reset,
    ReadyForQuery,
    SocketBackend,
    EndCommand,
    printtup_create_DR,
    NULL,
    NULL,
    libpq_report_param_status
};
```

对于内核代码而言，需要把函数的直接调用包装为port->protocol\_config->fn\_xxxx方式进行调用（例如：StartupPacketInitialize 修改为 port->protocol\_config->fn\_start）;在完全兼容既有内核的功能点的同时，增加了扩展其它网络协议的可能性。内核代码的修改点主要集中在 postmaster.cpp, postgres.cpp, postinit.cpp 中。

## 7.opengauss-server源码修改概述
```C++
src/common/backend/libpq/pqcomm.cpp                        |  25 ++++++++-----
  => 创建监听端口时，绑定网络协议扩展配置
src/common/backend/utils/error/elog.cpp                    |   9 +++--
  => 适配send_message_to_frontend
src/common/backend/utils/init/postinit.cpp                 |  11 +++---
  => 适配ClientAuthentication
src/common/backend/utils/misc/guc.cpp                      |  15 +++-----
  => 适配 report_param_status
src/gausskernel/process/postmaster/postmaster.cpp          | 227 +++++++++++++-----
  => 增加全局的协议扩展配置，定义opengauss默认网络协议实现，支持动态刷新监听的fd
src/gausskernel/process/tcop/dest.cpp                      |   2 +-
  => 适配 printtup_create_DR
src/gausskernel/process/tcop/postgres.cpp                  |  71 +++++++++++++++++++++--------------
  => 适配EndCommand,read_command,send_cancel_key, comm_reset, send_ready_for_query, process_command
src/gausskernel/process/threadpool/knl_thread.cpp          |   1 -
  =>移动ListenContext相关变量至g_instance中
src/gausskernel/process/threadpool/threadpool_listener.cpp |  6 ++++-
  => epoll中增加写事件，兼容由server端发起握手的数据库网络协议
src/gausskernel/process/threadpool/threadpool_worker.cpp   |  14 +++----
  => 适配send_ready_for_query, send_cancel_key
src/gausskernel/runtime/executor/spi.cpp                   |   9 ++++-
  => spi中兼容dolphin事务处理, 增加printtup需要的stmt参数透传
src/include/access/printtup.h                              |   1 +
  => DR_printtup增加target_list, 用于通过spi输出数据报文header到客户端
src/include/knl/knl_instance.h                             |  12 ++++++
  => 增加knl_g_listen_context结构体，用于运行时动态修改监听端口信息
src/include/knl/knl_thread.h                               |  7 +++---
  => 移动listen_context相关的属性至knl_instance.h
src/include/libpq/libpq-be.h                               |  28 ++++++++++++++
  => 定义ProtocolExtensionConfig接口，Port结构体中增加protocol_config属性
src/include/libpq/libpq.h                                  |   6 ++-
  => StreamServerPort中增加ProtocolExtensionConfig*参数
src/include/catalog/pg_authid.h                            |   6 +++++-
  => authid系统表中增加rolepasswordext，用于保存dolphin格式的密码

17 files changed, 291 insertions(+), 160 deletions(-)
```

## 8.opengauss适配网络协议接口方式

*对于opengauss内核代码，需要把 **default\_protocol\_config** 涉及的函数进行简单的适配，适配方法如下：*

***StreamConnection**：*

```C
protocol->fn_accept(serverFd, port) != STATUS_OK
```

**StreamClose:**

```C
protocol->fn_close(port->sock);
```

**pq_init:**

```C
if (port->protocol_config && port->protocol_config->fn_init) {
    port->protocol_config->fn_init(); 
} else {
    pq_init();
}
```

**StartupPacketInitialize**:

```C
int status = port->protocol_config->fn_start(port);
```

**ClientAuthentication:**

```C
port->protocol_config->fn_authenticate(port);
```

\*\*send\_message\_to_frontend: \*\*

```C
void send_message_to_frontend_ext(ErrorData* edata) {
    if (u_sess->proc_cxt.MyProcPort) {
        u_sess->proc_cxt.MyProcPort->protocol_config->fn_send_message(edata);
    }
}
```

**libpq\_send\_cancel_key:**

```C
Port* MyProcPort = u_sess->proc_cxt.MyProcPort;
if (MyProcPort && MyProcPort->protocol_config->fn_send_cancel_key) {
    MyProcPort->protocol_config->fn_send_cancel_key((int32)t_thrd.proc_cxt.MyPMChildSlot,
                                                            (int32)t_thrd.proc_cxt.MyCancelKey);
}

extern void libpq_send_cancel_key(int32 pid, int32 key);
void libpq_send_cancel_key(int32 pid, int32 key)
{
    StringInfoData buf;

pq_beginmessage(&buf, 'K');
    pq_sendint32(&buf, pid);
    pq_sendint32(&buf, key);
    pq_endmessage(&buf);
}
```

**pq\_comm\_reset:**

```C
Port* MyProcPort = u_sess->proc_cxt.MyProcPort;
if (MyProcPort && MyProcPort->protocol_config->fn_comm_reset) {
    MyProcPort->protocol_config->fn_comm_reset();
}
```

**ReadyForQuery:**

```C
session->proc_cxt.MyProcPort->protocol_config->fn_send_ready_for_query((CommandDest)t_thrd.postgres_cxt.whereToSendOutput);
```

**SocketBackend**

```C
result = u_sess->proc_cxt.MyProcPort->protocol_config->fn_read_command(inBuf);
```

**EndCommand**

```C
void EndCommandExt(const char *completionTag, CommandDest dest) {
    Port* MyProcPort = u_sess->proc_cxt.MyProcPort;
    if (MyProcPort) {
        MyProcPort->protocol_config->fn_end_command(completionTag, dest);
    }
}
```

**printtup\_create\_DR**

```C
u_sess->proc_cxt.MyProcPort->protocol_config->fn_printtup_create_DR(dest);
```

**libpq\_report\_param_status**

```C
/*
 * ReportGUCOption: if appropriate, transmit option value to frontend
 */
static void ReportGUCOption(struct config_generic* record)
{
    if (u_sess->utils_cxt.reporting_enabled && (record->flags & GUC_REPORT)) {
        Port* MyProcPort = u_sess->proc_cxt.MyProcPort;
        if (MyProcPort && MyProcPort->protocol_config->fn_report_param_status) {
            char* val = _ShowOption(record, false);
            (MyProcPort->protocol_config->fn_report_param_status)(record->name, val);
            pfree(val);
        }
    }
}

void libpq_report_param_status(const char *name, char *val)
{
    StringInfoData msgbuf;

pq_beginmessage(&msgbuf, 'S');
    pq_sendstring(&msgbuf, name);
    pq_sendstring(&msgbuf, val);
    pq_endmessage(&msgbuf);
}
```

openGauss/openGauss-server

内容风险标识

评论 (7)

openGauss/openGauss-server .gitee-modal { width: 500px !important; }

内容风险标识