ODBC驱动支持merge into和select走Batch模式 · Pull Request !43 · openGauss/openGauss-connector-odbc

【环境】
| 系统  | centos7   | x86_64 |
|-----|-----------|--------|
| 数据库 | opengauss | 5.0    |
| 驱动  | ODBC      | 5.0    |

【问题背景】
ODBC程序，用数组存放数据，然后遍历数组中的数据，执行merge操作。

```sql
merge into ... using ... on '='条件
when matched then update ...
when not matched then insert ...
```

该Merge语句，总共操作了多条数据，但在SQLRowCount()返回结果为1。即 MERGE 1。

【问题原因】
1、odbc驱动不支持merge操作的Batch模式。所以merge在odbc上执行，是非批量模式。
2、非batch模式，server端在执行merge时，由于merge条件是'='，即每次执行merge，满足条件的只有一条数据。
3、server端的执行过程，是批量从驱动取数据，然后在server端一条一条去执行。

【解决方案】
1、server端，在Batch模式下，支持delete,update,insert,select,merge操作，修改数量的累计功能(process_count）
2、ODBC驱动在判断SQL是否走Batch模式时，只支持了delete,update,insert。不支持merge和select。

结论：在驱动端，判断是否走Batch模式的逻辑时，新增select和merge。

demo程序：
MergeAffectedRowsTestSimple.cpp
```c
#include "common.h"
#include <cstring>
#include <cstdio>
#include <sqlext.h>
#include <stddef.h>
#include <chrono>
#include <iostream>

#define MERGE_ARRAY_SIZE 10

static bool merge_test() {
    const char *sql = "merge into t_table_for_test a using (select ? f_id, ? f_name, ? f_salary) nc"
                      " on (a.f_id = nc.f_id) "
                      " when matched then"
                      " update set a.f_name=nc.f_name, a.f_salary=nc.f_salary"
                      " when not matched then"
                      " insert(f_id, f_name, f_salary) values(nc.f_id, nc.f_name, nc.f_salary)";

SQLHSTMT stmt = SQL_NULL_HSTMT;

SQLRETURN ret = SQLAllocHandle(SQL_HANDLE_STMT, hdbc, &stmt);
    if (ret != SQL_SUCCESS) {
        printf("Alloc stmt error.\n");
        return false;
    }

SQLUSMALLINT status[MERGE_ARRAY_SIZE];
    SQLPOINTER processed;
    SQLSetStmtAttr(stmt, SQL_ATTR_PARAMSET_SIZE, (SQLPOINTER)MERGE_ARRAY_SIZE, 0);
    SQLSetStmtAttr(stmt, SQL_ATTR_PARAM_BIND_TYPE, (SQLPOINTER)sizeof(struct test_table_row), 0);
    SQLSetStmtAttr(stmt, SQL_ATTR_PARAMS_PROCESSED_PTR, (SQLPOINTER)&processed, 0);
    SQLSetStmtAttr(stmt, SQL_ATTR_PARAM_STATUS_PTR, (SQLPOINTER)status, 0);

ret = SQLPrepare(stmt, (SQLCHAR *)sql, SQL_NTS);
    if (ret != SQL_SUCCESS && ret != SQL_SUCCESS_WITH_INFO) {
        print_info(SQL_HANDLE_STMT, stmt); SQLFreeHandle(SQL_HANDLE_STMT, stmt);
        printf("Execute sql[%s] error.\n", sql);
        return false;
    }

struct test_table_row rows[MERGE_ARRAY_SIZE];

SQLBindParameter(stmt, 1, SQL_PARAM_INPUT, SQL_C_CHAR, SQL_VARCHAR, 0, 0, &rows[0].id, sizeof(rows[0].id.val), &rows[0].id.ind_or_len);
    SQLBindParameter(stmt, 2, SQL_PARAM_INPUT, SQL_C_CHAR, SQL_VARCHAR, 0, 0, &rows[0].name, sizeof(rows[0].name.val), &rows[0].name.ind_or_len);
    SQLBindParameter(stmt, 3, SQL_PARAM_INPUT, SQL_C_CHAR, SQL_VARCHAR, 0, 0, &rows[0].salary, sizeof(rows[0].salary.val), &rows[0].salary.ind_or_len);

for (int i = 0; i < MERGE_ARRAY_SIZE; i++) {
        sprintf((char *)rows[i].id.val, "%d", 27 + i); // 插入6条
        if (i > 5) sprintf((char *)rows[i].id.val, "%d", 27 + i - 5); // 更新5条
        sprintf((char *)rows[i].name.val, "%s", "jingtianlei");
        sprintf((char *)rows[i].salary.val, "%s", "99.99");

rows[i].id.ind_or_len = SQL_NTS;
        rows[i].name.ind_or_len = SQL_NTS;
        rows[i].salary.ind_or_len = SQL_NTS;
    }

ret = SQLExecute(stmt);
    if (ret != SQL_SUCCESS && ret != SQL_SUCCESS_WITH_INFO) {
        print_info(SQL_HANDLE_STMT, stmt);
        SQLFreeHandle(SQL_HANDLE_STMT, stmt);
        printf("Execute sql[%s] error.\n", sql);
        return false;
    }

SQLLEN rowc;
    ret = SQLRowCount(stmt, &rowc);
    printf("SQLRowCount return: ");
    print_return(ret);
    printf("%d lines effected.\n", rowc);

SQLEndTran(SQL_HANDLE_DBC, hdbc, SQL_COMMIT);
    SQLFreeHandle(SQL_HANDLE_STMT, stmt);

return true;
}

static bool merge_test0() {
    const char *sql = "merge into t_table_for_test a using (select ? f_id, ? f_name, ? f_salary) nc"
                      " on (a.f_id > ?) "
                      " when matched then"
                      " update set a.f_name=nc.f_name, a.f_salary=nc.f_salary"
                      " when not matched then"
                      " insert(f_id, f_name, f_salary) values(nc.f_id, nc.f_name, nc.f_salary)";

SQLHSTMT stmt = SQL_NULL_HSTMT;

SQLRETURN ret = SQLAllocHandle(SQL_HANDLE_STMT, hdbc, &stmt);
    if (ret != SQL_SUCCESS) {
        printf("Alloc stmt error.\n");
        return false;
    }

SQLUSMALLINT status[MERGE_ARRAY_SIZE];
    SQLPOINTER processed;
    SQLSetStmtAttr(stmt, SQL_ATTR_PARAMSET_SIZE, (SQLPOINTER)1, 0);      // 只执行一条
    SQLSetStmtAttr(stmt, SQL_ATTR_PARAM_BIND_TYPE, (SQLPOINTER)sizeof(struct test_table_row), 0);
    SQLSetStmtAttr(stmt, SQL_ATTR_PARAMS_PROCESSED_PTR, (SQLPOINTER)&processed, 0);
    SQLSetStmtAttr(stmt, SQL_ATTR_PARAM_STATUS_PTR, (SQLPOINTER)status, 0);

ret = SQLPrepare(stmt, (SQLCHAR *)sql, SQL_NTS);
    if (ret != SQL_SUCCESS && ret != SQL_SUCCESS_WITH_INFO) {
        print_info(SQL_HANDLE_STMT, stmt); SQLFreeHandle(SQL_HANDLE_STMT, stmt);
        printf("Execute sql[%s] error.\n", sql);
        return false;
    }

struct test_table_row rows[MERGE_ARRAY_SIZE];

SQLBindParameter(stmt, 1, SQL_PARAM_INPUT, SQL_C_CHAR, SQL_VARCHAR, 0, 0, &rows[0].id, sizeof(rows[0].id.val), &rows[0].id.ind_or_len);
    SQLBindParameter(stmt, 2, SQL_PARAM_INPUT, SQL_C_CHAR, SQL_VARCHAR, 0, 0, &rows[0].name, sizeof(rows[0].name.val), &rows[0].name.ind_or_len);
    SQLBindParameter(stmt, 3, SQL_PARAM_INPUT, SQL_C_CHAR, SQL_VARCHAR, 0, 0, &rows[0].salary, sizeof(rows[0].salary.val), &rows[0].salary.ind_or_len);
    SQLBindParameter(stmt, 4, SQL_PARAM_INPUT, SQL_C_CHAR, SQL_VARCHAR, 0, 0, &rows[0].id, sizeof(rows[0].id.val), &rows[0].id.ind_or_len);

sprintf((char *)rows[0].id.val, "%d", 27);
    sprintf((char *)rows[0].name.val, "%s", "jingtianlei");
    sprintf((char *)rows[0].salary.val, "%s", "99.99");

rows[0].id.ind_or_len = SQL_NTS;
    rows[0].name.ind_or_len = SQL_NTS;
    rows[0].salary.ind_or_len = SQL_NTS;

ret = SQLExecute(stmt);
    if (ret != SQL_SUCCESS && ret != SQL_SUCCESS_WITH_INFO) {
        print_info(SQL_HANDLE_STMT, stmt);
        SQLFreeHandle(SQL_HANDLE_STMT, stmt);
        printf("Execute sql[%s] error.\n", sql);
        return false;
    }

SQLLEN rowc;
    ret = SQLRowCount(stmt, &rowc);
    printf("SQLRowCount return: ");
    print_return(ret);
    printf("%d lines effected.\n", rowc);

SQLEndTran(SQL_HANDLE_DBC, hdbc, SQL_COMMIT);
    SQLFreeHandle(SQL_HANDLE_STMT, stmt);

return true;
}

int main(int argc, char *argv[]) {
    if (!open_connection("CLN_70_49", "tra_user", "Sgene1234!")) {
        printf("open connection failed.\n");
        return -1;
    }

default_drop_table();
    default_create_table();

merge_test();
    merge_test0();

if (argc == 2 && strcmp(argv[1], "-d") == 0) {
        default_drop_table();
    }
    
end:
    close_connection();

return 0;
}
```
common.cpp
```c
#include "common.h"
#include <stdio.h>
#include <sqlext.h>
#include <stddef.h>
#include <cstring>

SQLHENV henv;
SQLHDBC hdbc;

void print_status(SQLRETURN ret) {
    switch (ret) {
        case SQL_PARAM_SUCCESS: printf("SQL_PARAM_SUCCESS.\n"); break;
        case SQL_PARAM_SUCCESS_WITH_INFO: printf("SQL_PARAM_SUCCESS_WITH_INFO.\n"); break;
        case SQL_PARAM_ERROR: printf("SQL_PARAM_ERROR.\n"); break;
        case SQL_PARAM_UNUSED: printf("SQL_PARAM_ERROR.\n"); break;
        case SQL_PARAM_DIAG_UNAVAILABLE: printf("SQL_PARAM_DATA_AVAILABLE.\n"); break;
        default: printf("Other return[%d]\n", ret);
    }
}

void print_return(SQLRETURN ret) {
    switch (ret) {
        case SQL_SUCCESS: printf("SQL_SUCCESS.\n"); break;
        case SQL_SUCCESS_WITH_INFO: printf("SQL_SUCCESS_WITH_INFO.\n"); break;
        case SQL_ERROR: printf("SQL_ERROR.\n"); break;
        case SQL_NO_DATA: printf("SQL_NO_DATA.\n"); break;
        case SQL_NEED_DATA: printf("SQL_NEED_DATA.\n"); break;
        case SQL_INVALID_HANDLE: printf("SQL_INVALID_HANDLE.\n"); break;
        case SQL_PARAM_DATA_AVAILABLE: printf("SQL_PARAM_DATA_AVAILABLE.\n"); break;
        case SQL_PARAM_ERROR: printf("SQL_PARAM_ERROR.\n"); break;
        default: printf("Other return[%d]\n", ret);
    }
}

void print_info(SQLSMALLINT type, SQLHANDLE handle) {
    SQLCHAR sql_state[SQL_SQLSTATE_SIZE + 1];
    SQLINTEGER native_error_ptr = 0;
    SQLCHAR message_text[2048];
    SQLSMALLINT out_buffer_length;

SQLSMALLINT i_rec = 0;
    SQLRETURN ret;
    do {
        ret = SQLGetDiagRec(type,
                             handle,
                             ++i_rec,
                             sql_state,
                             &native_error_ptr,
                             message_text,
                             (SQLSMALLINT)sizeof(message_text),
                             &out_buffer_length);
        printf("SQLGetDiagRec return: ");
        print_return(ret);
        if (ret == SQL_NO_DATA) break;
        printf("state[%s] message[%s] error_code[%d]\n", sql_state, message_text, native_error_ptr);
    } while (ret == SQL_SUCCESS);

}

bool open_connection(const char *inst, const char *user, const char *passwd) {
        SQLRETURN ret = SQLAllocHandle(SQL_HANDLE_ENV, SQL_NULL_HANDLE, &henv);
        if (ret != SQL_SUCCESS) {
            printf("create env failed");
            return false;
        }

ret = SQLSetEnvAttr(henv, SQL_ATTR_ODBC_VERSION, (void*)SQL_OV_ODBC3, 0);
        if (ret != SQL_SUCCESS) {
            printf("create server handle failed");
            SQLFreeHandle(SQL_HANDLE_ENV, henv);
            return false;
        }

ret = SQLAllocHandle(SQL_HANDLE_DBC, henv, &hdbc);
        if (ret != SQL_SUCCESS) {
            printf("create error handler failed");
            SQLFreeHandle(SQL_HANDLE_ENV, henv);
            return false;
        }

ret = SQLSetConnectAttr(hdbc, SQL_ATTR_AUTOCOMMIT, (SQLPOINTER)SQL_AUTOCOMMIT_OFF, 0);
        if (ret != SQL_SUCCESS) {
            printf("alloc svr context handler failed");
            SQLFreeHandle(SQL_HANDLE_DBC, hdbc);
            SQLFreeHandle(SQL_HANDLE_ENV, henv);
            return false;
        }

ret = SQLConnect(hdbc, (SQLCHAR*)inst, SQL_NTS, (SQLCHAR*)user, SQL_NTS, (SQLCHAR*)passwd, SQL_NTS);
        if (ret != SQL_SUCCESS) {
            printf("connect to server failed");
            SQLFreeHandle(SQL_HANDLE_DBC, hdbc);
            SQLFreeHandle(SQL_HANDLE_ENV, henv);
            return false;
        }

return true;
}

bool close_connection() {
    SQLDisconnect(hdbc);
    SQLFreeHandle(SQL_HANDLE_DBC, hdbc);
    SQLFreeHandle(SQL_HANDLE_ENV, henv);
}

bool default_create_table() {
    const char *sql = "create table t_table_for_test(f_id integer primary key, f_name varchar(32), f_salary decimal(8, 2));";
    SQLHSTMT stmt = SQL_NULL_HSTMT;

SQLRETURN ret = SQLAllocHandle(SQL_HANDLE_STMT, hdbc, &stmt);
    if (ret != SQL_SUCCESS) {
        printf("Alloc stmt error.\n");
        return false;
    }

ret = SQLExecDirect(stmt, (SQLCHAR*)sql, SQL_NTS);
    if (ret != SQL_SUCCESS && ret != SQL_SUCCESS_WITH_INFO) {
        print_info(SQL_HANDLE_STMT, stmt);
        SQLFreeHandle(SQL_HANDLE_STMT, stmt);
        printf("Execute sql[%s] error.\n", sql);
        return false;
    }

if (ret == SQL_SUCCESS_WITH_INFO) {
        printf("create table success with info\n");
        print_info(SQL_HANDLE_STMT, stmt);
    }

if (ret == SQL_SUCCESS) {
        printf("Create table return SUCCESS.\n");
    }

SQLEndTran(SQL_HANDLE_DBC, hdbc, SQL_COMMIT);
    SQLFreeHandle(SQL_HANDLE_STMT, stmt);
    return true;
}

#define DEFAULT_ARRAY_SIZE 10

bool default_insert_data() {
    const char *sql = "insert into t_table_for_test values(?, ?, ?);";

SQLHSTMT stmt = SQL_NULL_HSTMT;

SQLRETURN ret = SQLAllocHandle(SQL_HANDLE_STMT, hdbc, &stmt);
    if (ret != SQL_SUCCESS) {
        printf("Alloc stmt error.\n");
        return false;
    }

SQLUSMALLINT status[DEFAULT_ARRAY_SIZE];
    SQLPOINTER processed;
    SQLSetStmtAttr(stmt, SQL_ATTR_PARAMSET_SIZE, (SQLPOINTER)DEFAULT_ARRAY_SIZE, 0);
    SQLSetStmtAttr(stmt, SQL_ATTR_PARAM_BIND_TYPE, (SQLPOINTER)sizeof(test_table_row), 0);
    SQLSetStmtAttr(stmt, SQL_ATTR_PARAMS_PROCESSED_PTR, (SQLPOINTER)&processed, 0);
    SQLSetStmtAttr(stmt, SQL_ATTR_PARAM_STATUS_PTR, (SQLPOINTER)status, 0);

struct test_table_row rows[DEFAULT_ARRAY_SIZE];

SQLBindParameter(stmt, 1, SQL_PARAM_INPUT, SQL_C_CHAR, SQL_INTEGER, 0, 0, &rows[0].id.val, sizeof(rows[0].id.val), &rows[0].id.ind_or_len);
    SQLBindParameter(stmt, 2, SQL_PARAM_INPUT, SQL_C_CHAR, SQL_VARCHAR, 0, 0, &rows[0].name.val, sizeof(rows[0].name.val), &rows[0].name.ind_or_len);
    SQLBindParameter(stmt, 3, SQL_PARAM_INPUT, SQL_C_CHAR, SQL_DECIMAL, 0, 0, &rows[0].salary.val, sizeof(rows[0].salary.val), &rows[0].salary.ind_or_len);

ret = SQLPrepare(stmt, (SQLCHAR *)sql, SQL_NTS);
    if (ret != SQL_SUCCESS && ret != SQL_SUCCESS_WITH_INFO) {
        print_info(SQL_HANDLE_STMT, stmt);
        SQLFreeHandle(SQL_HANDLE_STMT, stmt);
        printf("Execute sql[%s] error.\n", sql);
        return false;
    }

for (int i = 0; i < DEFAULT_ARRAY_SIZE; i++) {
        sprintf((char *)rows[i].id.val, "%d", i + 1);
        rows[i].id.ind_or_len = SQL_NTS;

if (i % 2) sprintf((char *)rows[i].name.val, "%s", "hanmeimei");
        else sprintf((char *)rows[i].name.val, "%s", "lilei");
        rows[i].name.ind_or_len = SQL_NTS;

sprintf((char *)rows[i].salary.val, "%.2f", 98.53);
        rows[i].salary.ind_or_len = SQL_NTS;
    }

ret = SQLExecute(stmt);
    if (ret != SQL_SUCCESS && ret != SQL_SUCCESS_WITH_INFO) {
        print_info(SQL_HANDLE_STMT, stmt);
        SQLFreeHandle(SQL_HANDLE_STMT, stmt);
        printf("Execute sql[%s] error.\n", sql);
        return false;
    }

if (ret == SQL_SUCCESS_WITH_INFO) {
        printf("create table success with info\n");
        print_info(SQL_HANDLE_STMT, stmt);
    }

if (ret == SQL_SUCCESS) {
        printf("insert data SUCCESS.\n");
    }

SQLLEN rowc;
    SQLRowCount(stmt, &rowc);

printf("%d lines effected.\n", rowc);

SQLEndTran(SQL_HANDLE_DBC, hdbc, SQL_COMMIT);
    SQLFreeHandle(SQL_HANDLE_STMT, stmt);

return true;
}

bool default_drop_table() {
    const char *sql = "drop table T_table_for_test";
    SQLHSTMT stmt = SQL_NULL_HSTMT;

SQLRETURN ret = SQLAllocHandle(SQL_HANDLE_STMT, hdbc, &stmt);
    if (ret != SQL_SUCCESS) {
        printf("Alloc stmt error.\n");
        return false;
    }

ret = SQLExecDirect(stmt, (SQLCHAR*)sql, SQL_NTS);
    if (ret != SQL_SUCCESS && ret != SQL_SUCCESS_WITH_INFO) {
        print_info(SQL_HANDLE_STMT, stmt);
        SQLFreeHandle(SQL_HANDLE_STMT, stmt);
        printf("Execute sql[%s] error.\n", sql);
        return false;
    }

if (ret == SQL_SUCCESS_WITH_INFO) {
        printf("drop table success with info\n");
        print_info(SQL_HANDLE_STMT, stmt);
    }

if (ret == SQL_SUCCESS) {
        printf("drop table return SUCCESS.\n");
    }

SQLEndTran(SQL_HANDLE_DBC, hdbc, SQL_COMMIT);
    SQLFreeHandle(SQL_HANDLE_STMT, stmt);

return true;
}
```

common.h
```c
#ifndef __COMMON_H__
#define __COMMON_H__

#include <sql.h>

extern SQLHENV henv;
extern SQLHDBC hdbc;

void print_status(SQLRETURN ret);
void print_return(SQLRETURN ret);
void print_info(SQLSMALLINT type, SQLHANDLE handle);
bool open_connection(const char *inst, const char *user, const char *passwd);
bool close_connection();

// default
bool default_create_table();
bool default_insert_data();
bool default_drop_table();

struct col_unit {
    SQLCHAR val[1024];
    SQLLEN  ind_or_len;
};

struct test_table_row {
    struct col_unit id;
    struct col_unit name;
    struct col_unit salary;
};

#endif
```

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

搜索帮助

openGauss/openGauss-connector-odbc