5. 程序开发示例

该附录中提供了几个 KingbaseES OCCI 的使用示例程序。

• 连接和断开数据库

• 执行批量插入

• 执行一条查询语句并获取结果

• 写入 BLOB/CLOB 数据

• 读取 BLOB/CLOB 数据

• 获取元信息

• 匿名块和游标

5.1. 连接和断开数据库

#include 
#include 
#include 
#include 
#include 
#include 
#include 

using namespace oracle::occi;
using namespace std;

/* 初始化OCCI环境资源，并创建连接 */
bool connect()
{
    /* 数据库的用户名和密码 */
    const string szUserName   = "system";
    const string szUserPwd    = "123456";
    /* KES的数据源名字，位于sys_service.conf中  */
    const string szConnection = "KingbaseES";

    if(connected)
        return true;

    /* 第一步，初始化OCCI的ENV环境 */
    pEnv = Environment::createEnvironment();
    try
    {
        /* 第二步，在ENV下创建到KES数据库的OCCI的连接对象 */
        pConn = pEnv->createConnection(szUserName, szUserPwd, szConnection);
    }
    catch (SQLException &ex)
    {
        /* 捕获OCCI异常并打印错误消息 */
        cout << ex.getMessage() << endl;
    }

    if (NULL == pConn)
    {
        cout << "Connection Failure" << endl;
        return false;
    }
    cout << "connect Success" << endl;
    connected = true;
    return true;
}

/* 断开连接，并释放OCCI的环境资源 */
bool disconnect()
{
    if(false == connected)
        return true;

    /* 断开数据库连接 */
    pEnv->terminateConnection(pConn);
    /* 销毁OCCI的环境资源 */
    Environment::terminateEnvironment(pEnv);
    delete pEnv;

    pConn = NULL;
    pEnv = NULL;
    cout << "disconnect Success" << endl;
    connected = false;
    return true;
}

5.2. 执行批量插入

/* 定义批量插入的行数 */
#define MAX_ARRAYSIZE 1000
/* 通过OCCI的循环迭代绑定功能做批量插入 */
void iterationDemo()
{
    Statement *pStmt = NULL;
    ResultSet *pRs = NULL;
    char msgBuf[100] = {0};
    struct timeval tvBegin, tvEnd;
    string strSqlSelect = "select * from gp_dupchk";
    string strSqCreate = "create table gp_dupchk(hash_id varchar(32) not null,"
                                                "file_nm varchar(64),"
                                                "cdr_item number(10),"
                                                "call_start_dt number(8) not null,"
                                                "primary key(hash_id) )";

    string strSqDrop = "drop table if exists gp_dupchk";
    string strSqlInsert = "insert into gp_dupchk values( :1, :2, :3, :4 )";
    unsigned int ret =0;

    print_bar((char*)__FUNCTION__);

    if(!connect())
    {
        cout<<__FUNCTION__<<" Failed"<createStatement();
        /* 关联一条SQL语句到该语句对象 */
        pStmt -> setSQL(strSqDrop);
        /* 执行非查询语句 */
        pStmt->executeUpdate();
        /* 关联另一条SQL语句到该语句对象 */
        pStmt -> setSQL(strSqCreate);
        ret = pStmt->executeUpdate();

        pStmt->setSQL(strSqlInsert);
        /* 设置迭代最大行数 */
        pStmt->setMaxIterations(MAX_ARRAYSIZE);
        /* 设置绑定参数的最大大小 */
        pStmt->setMaxParamSize(1, 32);
        pStmt->setMaxParamSize(2, 64);
        pStmt->setMaxParamSize(3, 10);
        pStmt->setMaxParamSize(4, 20);

        for(int i=0; isetString(1, msgBuf);
            pStmt->setString(2, "BBBBBB");
            pStmt->setInt(3, 1);
            pStmt->setNumber(4, Number(2));
            /* 添加到迭代器中 */
            pStmt->addIteration();
        }
        /*
         * 添加最后一行绑定参数
         * 不需要addInteration(),
         * 如果调用，将插入一个NULL行
         */
        pStmt->setString(1, "GGGGGG");
        pStmt->setString(2, "HHHHHH");
        pStmt->setInt(3, 7);
        pStmt->setNumber(4, Number(8));

        gettimeofday(&tvBegin, NULL);
        /* 执行批量插入 */
        ret = pStmt->executeUpdate();
        /* 获取实际影响行数 */
        if(pStmt->getUpdateCount()!= MAX_ARRAYSIZE ){
            cout << "batch insert failed!
                    Expect:"<< MAX_ARRAYSIZE << "
                    but getUpdateCount: " <<  pStmt->getUpdateCount() << endl;
        }else{
            cout << "batch insert success" << endl;
        }


        gettimeofday(&tvEnd, NULL);
        printf("batch insert time: [%ld]\n",
                   (tvEnd.tv_sec - tvBegin.tv_sec) * 1000 +
                   (tvEnd.tv_usec - tvBegin.tv_usec) / 1000);
        cout << "insert getUpdateCount(): " << pStmt->getUpdateCount() << endl;
        /* 释放语句对象资源 */
        pConn -> terminateStatement(pStmt);
        pStmt = NULL;
    }
    catch ( SQLException &ex)
    {
        cout << "ErrorCode: " << ex.getErrorCode() << endl;
        cout << ex.getMessage() << endl;
        cout << "insert getUpdateCount(): " << pStmt->getUpdateCount() << endl;
    }

    disconnect();
    cout << "iterationDemo Success" << endl;
}

5.3. 执行一条查询语句并获取结果

/* 通过OCCI执行查询并获取返回结果集 */
void selectDemo()
{
    Statement *pStmt = NULL;
    ResultSet *pRs = NULL;

    string strSqlSelect = "select * from t_occi_selectDemo";
    string strSqCreate = "create table t_occi_selectDemo(c_id int,
                                                         c_name varchar(40),
                                                         c_number numeric(6,2),
                                                         c_number8 numeric(15,8))";
    string strSqDrop = "drop table IF EXISTS t_occi_selectDemo";

    string strSqlinsert1 = "insert into t_occi_selectDemo
                                        values(1,'Jack',11.11,33.33333333)";
    string strSqlinsert2 = "insert into t_occi_selectDemo
                                        values(2,'Tom',22.22,44.44444444)";

    print_bar((char*)__FUNCTION__);

    if(!connect())
    {
            cout<<__FUNCTION__<<" Failed"<createStatement();
       pStmt -> setSQL(strSqDrop);
       pStmt->executeUpdate();
       pStmt -> setSQL(strSqCreate);
       pStmt->executeUpdate();

       pStmt->executeUpdate(strSqlinsert1);
       pStmt->executeUpdate(strSqlinsert2);

       /* 关联查询语句 */
       pStmt -> setSQL(strSqlSelect);
       /* 执行查询，并获取返回结果集对象 */
       pRs = pStmt->executeQuery();
       /* 遍历结果集 */
       while ( pRs -> next())
       {
               /* 获取结果集中每个字段的值 */
               cout << pRs->getInt(1) << " "
                                      << pRs->getString(2) << " " << pRs->getFloat(3)
                                      << " " << pRs->getFloat(4) < closeResultSet(pRs);
       pRs = NULL;

       /* 清除表和数据 */
       pStmt->execute(strSqDrop);

       pConn -> terminateStatement(pStmt);
       pStmt = NULL;
    }
    catch ( SQLException &ex)
    {
       cout << ex.getMessage() << endl;
    }

    disconnect();
    cout << "selectDemo Success" << endl;
}

5.4. 写入 BLOB/CLOB 数据

/* 通过OCCI写入大对象 */
void lobWriteDemo()
{
    Statement *pStmt = NULL;
    ResultSet *pRs = NULL;
    unsigned int amt, offset, len, act_read_count;
    int not_read_count;
    unsigned char buff[READ_BUFSIZE];
    FILE *fp = NULL;

    /* 通过for update语句获取大对象 */
    string strSqlSelect = "select * from t_lob for update";
    string strSqCreate = "create table t_lob(b blob, c1 clob, c2 nclob)";
    string strSqDrop = "DROP TABLE IF EXISTS t_lob";
    string strSqlinsert = "INSERT INTO T_LOB
                                       VALUES(EMPTY_BLOB(),
                                       EMPTY_CLOB(),
                                       EMPTY_NCLOB())";

    unsigned char * cbuff = (unsigned char *)malloc(READ_BUFSIZE * 512 + 1);
    if(NULL == cbuff)
    {
        printf("unsigned char * cbuff = malloc(READ_BUFSIZE * 512 + 1);
                                                            failed\n");
        return;
    }

    print_bar((char*)__FUNCTION__);

    if(!connect())
    {
        cout<<__FUNCTION__<<" Failed"<createStatement();

       pStmt -> setSQL(strSqDrop);
       pStmt->executeUpdate();

       pStmt -> setSQL(strSqCreate);
       pStmt->executeUpdate();

       pStmt -> setSQL(strSqlinsert);
       pStmt->executeUpdate();
       /* 通过select for update获取大对象并锁定该行 */
       pStmt -> setSQL(strSqlSelect);
       pRs = pStmt->executeQuery();

       while (pRs->next())
       {
            /* 从结果集中获取OCCI的Blob对象 */
            Blob blob = pRs->getBlob(1);
            if(blob.isNull())
            {
                cout << "Null Blob" << endl;
            }
            else
            {
                /* 以读写方式打开Blob对象 */
                blob.open(OCCI_LOB_READWRITE);
                offset = 1;
                not_read_count = 1548;
                /* 打开本地磁盘文件 */
                fp = fopen("blob_in", "rb");
                /*
                 * 循环读取文件中的数据，
                 * 每次最多读取READ_BUFSIZE的块，
                 * 把每次读到的数据写入blob对象中
                 */
                while (not_read_count > 0)
                {
                    not_read_count -= READ_BUFSIZE;

                    if (not_read_count < 0)
                    {
                        act_read_count = READ_BUFSIZE + not_read_count;
                    }
                    else
                    {
                        act_read_count = READ_BUFSIZE;
                    }

                    memset(buff, 0, act_read_count);
                    fread(buff, 1, act_read_count, fp);

                    /*
                     * 写入到blob大对象中
                     * amt是本次期望写入的大小
                     * buff是待写入的数据
                     * act_read_count是buff的大小
                     * offset是写入的偏移量
                     */
                    amt = act_read_count;
                    act_read_count = blob.writeChunk(amt, buff, act_read_count, offset);
                    if (act_read_count > 0)
                    {
                        offset += amt;
                    }
                }

                /* 获取blob的数据长度 */
                len = blob.length();
                /* 关闭blob对象 */
                blob.close();
                fclose(fp);
                cout <<"write " << len << " into blob" << endl;
            }

            /* Clob的写入操作，和Blob类似 */
            Clob clob1 = pRs->getClob(2);
            if(clob1.isNull())
            {
                cout << "Null Clob" << endl;
            }
            else
            {
                clob1.open(OCCI_LOB_READWRITE);
                offset = 1;
                not_read_count = 1483407;
                fp = fopen("clob1_in.txt", "r");

                while (not_read_count > 0)
                {
                    if (not_read_count > READ_BUFSIZE * 512)
                        act_read_count = READ_BUFSIZE * 512;
                    else
                        act_read_count = not_read_count;

                    memset(cbuff, 0, READ_BUFSIZE * 512 + 1);
                    fread(cbuff, 1, act_read_count, fp);
                    cbuff[act_read_count] = '\0';

                    /* 将其写入定位器中 */
                    amt = act_read_count;
                    clob1.writeChunk(amt, cbuff, act_read_count + 1, offset);

                    offset += amt;
                    not_read_count -= act_read_count;
                }

                len = clob1.length();
                clob1.close();
                fclose(fp);
                cout <<"write " << len << " into clob1" << endl;
            }

            Clob clob2 = pRs->getClob(3);
            if(clob2.isNull())
            {
                cout << "Null Clob" << endl;
            }
            else
            {
                offset = 1;

#if defined(WIN32)
                not_read_count = 4477;
                fp = fopen("clob2_in_gb2312.txt", "r");
#else
                not_read_count = 4500;
                fp = fopen("clob2_in.txt", "r");
#endif

                while (not_read_count > 0)
                {
                    if (not_read_count > READ_BUFSIZE)
                        act_read_count = READ_BUFSIZE;
                    else
                        act_read_count = not_read_count;

                    memset(cbuff, 0, READ_BUFSIZE + 1);
                    fread(cbuff, 1, act_read_count, fp);

                    cbuff[act_read_count] = '\0';

                    /* 将其写入定位器中 */
                    amt = act_read_count;
                    clob2.write(amt, cbuff, act_read_count + 1, offset);

                    offset += amt;
                    not_read_count -= act_read_count;
                }

                len = clob2.length();
                fclose(fp);
                cout <<"write " << len << " into clob2" << endl;
            }
       }

       pStmt -> closeResultSet(pRs);
       pRs = NULL;

       pConn -> terminateStatement(pStmt);
       pStmt = NULL;
    }
    catch ( SQLException &ex)
    {
       cout << ex.getMessage() << endl;
       cout << "lobWriteDemo caught exception!!!" << endl;
    }

    disconnect();
    free(cbuff);
    cout << "lobWriteDemo Success" << endl;
}

5.5. 读取 BLOB/CLOB 数据

/* 通过OCCI读取大对象 */
void lobReadDemo()
{
    Statement *pStmt = NULL;
    ResultSet *pRs = NULL;
    unsigned int amt, offset, len, act_read_count;
    unsigned char buff[READ_BUFSIZE];
    FILE *fp = NULL;

    string strSqlSelect = "select * from t_lob";

    print_bar((char*)__FUNCTION__);

    if(!connect())
    {
        cout<<__FUNCTION__<<" Failed"<createStatement();

       pStmt -> setSQL(strSqlSelect);
       pRs = pStmt->executeQuery();

       while (pRs->next())
       {
            Blob blob = pRs->getBlob(1);
            if(blob.isNull())
            {
                cout << "Null Blob" << endl;
            }
            else
            {
                /* 以只读方式打开大对象 */
                blob.open(OCCI_LOB_READONLY);
                /* 获取大对象长度 */
                len = blob.length();
                cout << "blob: len = " << len << endl;
                amt = READ_BUFSIZE;

                /* 将BLOB数据读取到文件中 */
                offset = 1;
                fp = fopen("blob_out", "wb");
                /* 分块循环读取大对象数据 */
                while(1)
                {
                    memset((dvoid *)buff, 0, READ_BUFSIZE);
                    /* 从大对象中偏移读取一块指定大小的数据 */
                    act_read_count = blob.read(amt, buff, READ_BUFSIZE, offset);
                    if(act_read_count > 0)
                    {
                        fwrite(buff, act_read_count, 1, fp);
                        offset += act_read_count;
                    }
                    else
                    {
                        break;
                    }
                }
                blob.close();
                fclose(fp);
            }

            /* Clob 的读取，和Blob类似 */
            Clob clob1 = pRs->getClob(2);
            if(clob1.isNull())
            {
                cout << "Null Clob" << endl;
            }
            else
            {
                clob1.open(OCCI_LOB_READONLY);
                len = clob1.length();
                cout << "clob1: len = " << len << endl;
                amt = READ_BUFSIZE - 1;

                /* 将CLOB数据读取到文件中 */
                offset = 1;
                fp = fopen("clob1_out.txt", "wb");

                while(1)
                {
                    memset((dvoid *)buff, 0, READ_BUFSIZE);
                    act_read_count = clob1.read(amt, buff, READ_BUFSIZE, offset);
                    if(act_read_count > 0)
                    {
                        fwrite(buff, act_read_count, 1, fp);
                        offset += act_read_count;
                    }
                    else
                    {
                        break;
                    }
                }
                clob1.close();
                fclose(fp);
            }

            Clob clob2 = pRs->getClob(3);
            if(clob2.isNull())
            {
                cout << "Null Clob" << endl;
            }
            else
            {
                clob2.open(OCCI_LOB_READONLY);
                len = clob2.length();
                cout << "clob2: len = " << len << endl;
                amt = READ_BUFSIZE - 1;

                /* 将CLOB数据读取到文件中 */
                offset = 1;
                fp = fopen("clob2_out.txt", "wb");

                while(1)
                {
                    memset((dvoid *)buff, 0, READ_BUFSIZE);
                    act_read_count = clob2.read(amt, buff, READ_BUFSIZE, offset);
                    if(act_read_count > 0)
                    {
                        fwrite(buff, act_read_count, 1, fp);
                        offset += act_read_count;
                    }
                    else
                    {
                        break;
                    }
                }
                clob2.close();
                fclose(fp);
            }
       }

       pStmt -> closeResultSet(pRs);
       pRs = NULL;

       pConn -> terminateStatement(pStmt);
       pStmt = NULL;
    }
    catch ( SQLException &ex)
    {
       cout << ex.getMessage() << endl;
       cout << "lobReadDemo caught exception!!!" << endl;
    }

    disconnect();
    cout << "lobReadDemo Success" << endl;
}

5.6. 获取元信息

void MetaDataDemo()
{
    Statement *pStmt = NULL;

    string strSqCreate = "create table t_occi_MeTaDataDemo(
                                       c_id int,c_name varchar(20),
                                       c_date date,c_timestamp timestamp)";
    string strSqDrop = "drop table if EXISTS t_occi_MeTaDataDemo";

    print_bar((char*)__FUNCTION__);

    if(!connect())
    {
        cout<<__FUNCTION__<<" Failed"<createStatement();
       pStmt -> setSQL(strSqDrop);
       pStmt->executeUpdate();
       pStmt -> setSQL(strSqCreate);
       pStmt->executeUpdate();

       /* 获取数据库中指定表的元信息 */
       MetaData meta = pConn->getMetaData(
                           "T_OCCI_METADATADEMO",MetaData::PTYPE_TABLE);
       /* 获取对象类型 */
       if(meta.getInt(MetaData::ATTR_PTYPE) ==MetaData::PTYPE_TABLE)
       {
           cout << "it is table" << endl;
           /* 获取表的字段属性信息 */
           vector tab_col_list = meta.getVector(
                                       MetaData::ATTR_LIST_COLUMNS);
            /* 遍历获取每一个字段的元信息 */
            std::vector::iterator p;
            unsigned int i;
            p = tab_col_list.begin();

            for(i = 0; i < tab_col_list.size(); i++)
            {
                /* 获取子弹的名字和类型 */
                cout << "ATTR_NAME " << i << " "
                            << p->getString(MetaData::ATTR_NAME);
                cout << " " << p->getInt(MetaData::ATTR_DATA_TYPE) << endl;
                p++;
            }
       }
       else
           cout << "it is not table" << endl;

       pStmt->execute(strSqDrop);

       pConn -> terminateStatement(pStmt);
       pStmt = NULL;
    }
    catch ( SQLException &ex)
    {
        cout << "error:" << ex.getMessage() << endl;
    }

    disconnect();
    cout << "MetaDataDemo Success" << endl;
}

5.7. 匿名块和游标

void PackageCursorNoBeginDemo()
{
    print_bar((char*)__FUNCTION__);

    if(!connect())
    {
        cout<<__FUNCTION__<<" Failed"< a; "
        "c := cur1;"
        "end;"
        "END;";

    OraText * sql_out_refcursor = (OraText*)"begin public.cursor_pkg.fa_out_refcursor
                                                         (:int_,:stmt_recursor);end;";

    Statement *stmt = pConn->createStatement();
    try{
        /* 主备表，存储过程，数据 */
        cout << "Create table and proc :" << sqlPackageHead << endl;
        stmt->setSQL((char*)sqldropTable);
        stmt->execute();
        stmt->setSQL((char*)sqlcreateTable);
        stmt->execute();
        stmt->setSQL((char*)sqlInsert);
        stmt->execute();
        stmt->setSQL((char*)sqlPackageHead);
        stmt->execute();
        stmt->setSQL((char*)sqlPackageBody);
        stmt->execute();

        /* 绑定INOUT参数，执行匿名块 */
        cout << "Executing the block :" << sql_out_refcursor << endl;
        stmt->setSQL((char*)sql_out_refcursor);
        stmt->setInt (1, 1);
        stmt->registerOutParam (2, OCCICURSOR);
        int updateCount = stmt->executeUpdate ();
        cout << "Update Count:" << updateCount << endl;
        /* 获取匿名块返回的游标 */
        ResultSet* rs_cursor = stmt->getCursor (2);
        /* 遍历游标中的结果集数据 */
        while(rs_cursor->next())
        {
            cout << "Printing the OUT cursor:" << endl;
            cout << "Cursor's Col1: " << rs_cursor->getInt(1) << endl;
            cout << "Cursor's Col2: " << rs_cursor->getString(2) << endl;
        }
        stmt->closeResultSet(rs_cursor);

        pConn->terminateStatement (stmt);
        cout << "occiproc - done" << endl;

        if(!disconnect())
        {
            cout<<__FUNCTION__<<" Failed"<