PostgreSQL: support binary data PostgreSQl: add database pool

znone

2021-02-26 a2b2faa6019572388248617d0ac740bde95feb74

PostgreSQL: support binary data
PostgreSQl: add database pool

 README.md

@@ -413,6 +413,8 @@
| real | float |
| DOUBLE | double |
| text | const char*<br>std::string |
| bytea | qtl::const_blob_data<br>std::vector<uint8_t> |
| oid | qtl::postgres::large_object |
| date | qtl::postgres::date |
| timestamp | qtl::postgres::timestamp |
| interval | qtl::postgres::interval |
@@ -428,6 +430,8 @@
| real | float |
| DOUBLE | double |
| text | char[N]<br>std::array&lt;char, N&gt;<br>std::string |
| bytea | qtl::const_blob_data<br>qtl::blob_data<br>std::vector<uint8_t> |
| oid | qtl::postgres::large_object |
| date | qtl::postgres::date |
| timestamp | qtl::postgres::timestamp |
| interval | qtl::postgres::interval |
@@ -448,7 +452,9 @@

Third-party libraries for compiling test cases need to be downloaded separately. In addition to database-related libraries, test cases use a test framework[CppTest](https://sourceforge.net/projects/cpptest/ "CppTest")。

The MySQL database used in the test case is as follows:
The database used in the test case is as follows:

### MySQL
```SQL
CREATE TABLE test (
  ID int NOT NULL AUTO_INCREMENT,
@@ -465,3 +471,37 @@
  PRIMARY KEY (ID)
);
```

### PostgreSQL
```SQL
DROP TABLE IF EXISTS test;
CREATE TABLE test (
  id int4 NOT NULL GENERATED BY DEFAULT AS IDENTITY (
INCREMENT 1
MINVALUE  1
MAXVALUE 2147483647
START 1
),
  name varchar(255) COLLATE default,
  createtime timestamp(6)
)
;

ALTER TABLE test ADD CONSTRAINT test_pkey PRIMARY KEY ("id");

DROP TABLE IF EXISTS test_blob;
CREATE TABLE test_blob (
  id int4 NOT NULL GENERATED BY DEFAULT AS IDENTITY (
INCREMENT 1
MINVALUE  1
MAXVALUE 2147483647
START 1
),
  filename varchar(255) COLLATE default NOT NULL,
  md5 bytea,
  content oid
)
;

ALTER TABLE test_blob ADD CONSTRAINT test_blob_pkey PRIMARY KEY ("id");
```

 README_CN.md

@@ -413,6 +413,8 @@
| real | float |
| DOUBLE | double |
| text | const char*<br>std::string |
| bytea | qtl::const_blob_data<br>std::vector<uint8_t> |
| oid | qtl::postgres::large_object |
| date | qtl::postgres::date |
| timestamp | qtl::postgres::timestamp |
| interval | qtl::postgres::interval |
@@ -428,6 +430,8 @@
| real | float |
| DOUBLE | double |
| text | char[N]<br>std::array&lt;char, N&gt;<br>std::string |
| bytea | qtl::const_blob_data<br>qtl::blob_data<br>std::vector<uint8_t> |
| oid | qtl::postgres::large_object |
| date | qtl::postgres::date |
| timestamp | qtl::postgres::timestamp |
| interval | qtl::postgres::interval |
@@ -448,7 +452,9 @@

编译测试用例的第三方库需要另外下载。除了数据库相关的库外，测试用例用到了测试框架[CppTest](https://sourceforge.net/projects/cpptest/ "CppTest")。

测试用例所用的MySQL数据库如下：
测试用例所用的数据库如下：

### MySQL
```SQL
CREATE TABLE test (
  ID int NOT NULL AUTO_INCREMENT,
@@ -466,4 +472,39 @@
);
```

### PostgreSQL
```SQL
DROP TABLE IF EXISTS test;
CREATE TABLE test (
  id int4 NOT NULL GENERATED BY DEFAULT AS IDENTITY (
INCREMENT 1
MINVALUE  1
MAXVALUE 2147483647
START 1
),
  name varchar(255) COLLATE default,
  createtime timestamp(6)
)
;

ALTER TABLE test ADD CONSTRAINT test_pkey PRIMARY KEY ("id");

DROP TABLE IF EXISTS test_blob;
CREATE TABLE test_blob (
  id int4 NOT NULL GENERATED BY DEFAULT AS IDENTITY (
INCREMENT 1
MINVALUE  1
MAXVALUE 2147483647
START 1
),
  filename varchar(255) COLLATE default NOT NULL,
  md5 bytea,
  content oid
)
;

ALTER TABLE test_blob ADD CONSTRAINT test_blob_pkey PRIMARY KEY ("id");
```


测试用例在 Visual Studio 2013 和 GCC 4.8 下测试通过。

 include/qtl_common.hpp

@@ -1158,6 +1158,14 @@
        overflow();
    }

    void swap(blobbuf& other)
    {
        std::swap(m_buf, other.m_buf);
        std::swap(m_size, other.m_size);
        std::swap(m_pos, other.m_pos);

        std::streambuf::swap(other);
    }

protected:
    virtual pos_type seekoff(off_type off, std::ios_base::seekdir dir,

 include/qtl_mysql.hpp

@@ -245,6 +245,13 @@
        init_buffer(mode);
    }

    void swap(blobbuf& other)
    {
        std::swap(m_stmt, other.m_stmt);
        std::swap(m_binder, other.m_binder);
        std::swap(m_field, other.m_field);
        qtl::blobbuf::swap(other);
    }

private:
    MYSQL_STMT* m_stmt;

 include/qtl_postgres.hpp

@@ -11,11 +11,13 @@
#include <sstream>
#include <chrono>
#include <algorithm>
#include <assert.h>
#include "qtl_common.hpp"

#define FRONTEND

#include <libpq-fe.h>
#include <libpq/libpq-fs.h>
#include <pgtypes_error.h>
#include <pgtypes_interval.h>
#include <pgtypes_timestamp.h>
@@ -52,6 +54,12 @@
#ifdef printf
#undef printf
#endif
#ifdef rename
#undef rename
#endif
#ifdef unlink
#undef unlink
#endif

namespace qtl
{
@@ -63,7 +71,7 @@

    inline int16_t ntoh(int16_t v)
    {
        return ntohs(v);
        return static_cast<int16_t>(ntohs(v));
    }
    inline uint16_t ntoh(uint16_t v)
    {
@@ -71,7 +79,7 @@
    }
    inline int32_t ntoh(int32_t v)
    {
        return ntohl(v);
        return static_cast<int32_t>(ntohl(v));
    }
    inline uint32_t ntoh(uint32_t v)
    {
@@ -99,7 +107,7 @@

    inline int16_t hton(int16_t v)
    {
        return htons(v);
        return static_cast<int16_t>(htons(v));
    }
    inline uint16_t hton(uint16_t v)
    {
@@ -107,7 +115,7 @@
    }
    inline int32_t hton(int32_t v)
    {
        return htonl(v);
        return static_cast<int32_t>(htonl(v));
    }
    inline uint32_t hton(uint32_t v)
    {
@@ -532,6 +540,150 @@
    return a.compare(b) != 0;
}

class large_object : public qtl::blobbuf
{
public:
    large_object() : m_conn(nullptr), m_id(InvalidOid), m_fd(-1) { }
    large_object(PGconn* conn, Oid loid, std::ios_base::openmode mode)
    {
        open(conn, loid, mode);
    }
    large_object(const large_object&) = delete;
    large_object(large_object&& src)
    {
        swap(src);
        src.m_conn = nullptr;
        src.m_fd = -1;
    }
    ~large_object()
    {
        close();
    }

    static large_object create(PGconn* conn, Oid loid = InvalidOid)
    {
        Oid oid = lo_create(conn, loid);
        if (oid == InvalidOid)
            throw error(conn);
        return large_object(conn, oid, std::ios::in|std::ios::out|std::ios::binary);
    }
    static large_object load(PGconn* conn, const char* filename, Oid loid = InvalidOid)
    {
        Oid oid = lo_import_with_oid(conn, filename, loid);
        if (oid == InvalidOid)
            throw error(conn);
        return large_object(conn, oid, std::ios::in | std::ios::out | std::ios::binary);
    }
    void save(const char* filename) const
    {
        if (lo_export(m_conn, m_id, filename) < 0)
            throw error(m_conn);
    }

    void unlink()
    {
        close();
        if (lo_unlink(m_conn, m_id) < 0)
            throw error(m_conn);
    }

    large_object& operator=(const large_object&) = delete;
    large_object& operator=(large_object&& src)
    {
        if (this != &src)
        {
            swap(src);
            src.close();
        }
        return *this;
    }
    bool is_open() const { return m_fd >= 0; }
    Oid oid() const { return m_id; }

    void open(PGconn* conn, Oid loid, std::ios_base::openmode mode)
    {
        int lomode = 0;
        if (mode&std::ios_base::in)
            lomode |= INV_READ;
        if (mode&std::ios_base::out)
            lomode |= INV_WRITE;
        m_conn = conn;
        m_id = loid;
        m_fd = lo_open(m_conn, loid, lomode);
        if (m_fd < 0)
            throw error(m_conn);

        m_size = size();
        init_buffer(mode);
        if (mode&std::ios_base::trunc)
        {
            if (lo_truncate(m_conn, m_fd, 0) < 0)
                throw error(m_conn);
        }
    }

    void close()
    {
        if (m_fd >= 0)
        {
            overflow();
            if (lo_close(m_conn, m_fd) < 0)
                throw error(m_conn);
            m_fd = -1;
        }
    }

    void flush()
    {
        if (m_fd >= 0)
            overflow();
    }

    size_t size() const
    {
        pg_int64 size = 0;
        if (m_fd >= 0)
        {
            pg_int64 org = lo_tell64(m_conn, m_fd);
            size = lo_lseek64(m_conn, m_fd, 0, SEEK_END);
            lo_lseek64(m_conn, m_fd, org, SEEK_SET);
        }
        return size;
    }

    void resize(size_t n)
    {
        if (m_fd >= 0 && lo_truncate64(m_conn, m_fd, n) < 0)
            throw error(m_conn);
    }

    void swap(large_object& other)
    {
        std::swap(m_conn, other.m_conn);
        std::swap(m_id, other.m_id);
        std::swap(m_fd, other.m_fd);
        qtl::blobbuf::swap(other);
    }

protected:
    enum { default_buffer_size = 4096 };

    virtual bool read_blob(char* buffer, off_type& count, pos_type position) override
    {
        return lo_lseek64(m_conn, m_fd, position, SEEK_SET) >= 0 && lo_read(m_conn, m_fd, buffer, count) > 0;
    }
    virtual void write_blob(const char* buffer, size_t count) override
    {
        if (lo_write(m_conn, m_fd, buffer, count) < 0)
            throw error(m_conn);
    }

private:
    PGconn* m_conn;
    Oid m_id;
    int m_fd;
};

/*
    template<typename T>
    struct oid_traits
@@ -696,6 +848,61 @@
    }
};

template<> struct object_traits<qtl::const_blob_data> : public base_object_traits<qtl::const_blob_data, BYTEAOID>
{
    static value_type get(const char* data, size_t n)
    {
        return qtl::const_blob_data(data, n);
    }
    static std::pair<const char*, size_t> data(const qtl::const_blob_data& v, std::vector<char>& data)
    {
        assert(v.size <= UINT32_MAX);
        return std::make_pair(static_cast<const char*>(v.data), v.size);
    }
};

template<> struct object_traits<qtl::blob_data> : public base_object_traits<qtl::blob_data, BYTEAOID>
{
    static void get(qtl::blob_data& value, const char* data, size_t n)
    {
        if (value.size < n)
            throw std::out_of_range("no enough buffer to receive blob data.");
        memcpy(value.data, data, n);
    }
    static std::pair<const char*, size_t> data(const qtl::blob_data& v, std::vector<char>& data)
    {
        assert(v.size <= UINT32_MAX);
        return std::make_pair(static_cast<char*>(v.data), v.size);
    }
};

template<> struct object_traits<std::vector<uint8_t>> : public base_object_traits<std::vector<uint8_t>, BYTEAOID>
{
    static value_type get(const char* data, size_t n)
    {
        const uint8_t* begin = reinterpret_cast<const uint8_t*>(data);
        return std::vector<uint8_t>(begin, begin+n);
    }
    static std::pair<const char*, size_t> data(const std::vector<uint8_t>& v, std::vector<char>& data)
    {
        assert(v.size() <= UINT32_MAX);
        return std::make_pair(reinterpret_cast<const char*>(v.data()), v.size());
    }
};

template<> struct object_traits<large_object> : public base_object_traits<large_object, OIDOID>
{
    static value_type get(PGconn* conn, const char* data, size_t n)
    {
        Oid oid = object_traits<int32_t>::get(data, n);
        return large_object(conn, oid, std::ios::in | std::ios::out | std::ios::binary);
    }
    static std::pair<const char*, size_t> data(const large_object& v, std::vector<char>& data)
    {
        return object_traits<int32_t>::data(v.oid(), data);
    }
};

struct binder
{
    binder() = default;
@@ -725,6 +932,22 @@
            throw std::bad_cast();

        return object_traits<T>::get(m_value, m_length);
    }
    template<typename T>
    T get(PGconn* conn)
    {
        if (!object_traits<T>::is_match(m_type))
            throw std::bad_cast();

        return object_traits<T>::get(conn, m_value, m_length);
    }
    template<typename T>
    void get(T& v)
    {
        if (!object_traits<T>::is_match(m_type))
            throw std::bad_cast();

        return object_traits<T>::get(v, m_value, m_length);
    }

    void bind(std::nullptr_t)
@@ -983,6 +1206,15 @@
        }
    }

    void bind_field(size_t index, large_object&& value)
    {
        value = m_binders[index].get<large_object>(m_conn);
    }
    void bind_field(size_t index, blob_data&& value)
    {
        m_binders[index].get(value);
    }

protected:
    PGconn* m_conn;
    result m_res;
@@ -1089,7 +1321,7 @@
        if (!PQsetSingleRowMode(m_conn))
            throw error(m_conn);
        m_res = PQgetResult(m_conn);
        verify_error<PGRES_COMMAND_OK, PGRES_SINGLE_TUPLE>();
        verify_error<PGRES_COMMAND_OK, PGRES_SINGLE_TUPLE, PGRES_TUPLES_OK>();
    }

    template<typename Types>

 include/qtl_postgres_pool.hpp

New file
@@ -0,0 +1,45 @@
#ifndef _QTL_MYSQL_POOL_H_
#define _QTL_MYSQL_POOL_H_

#include "qtl_database_pool.hpp"
#include "qtl_postgres.hpp"

namespace qtl
{

namespace postgres
{

class database_pool : public qtl::database_pool<database>
{
public:
    database_pool() : m_port(0) { }
    virtual ~database_pool() { }
    virtual database* new_database() throw() override
    {
        database* db=new database;
        if(!db->open(m_host.data(), m_user.data(), m_password.data(), m_port, m_database.data()))
        {
            delete db;
            db=NULL;
        }
        else
        {
            PQsetClientEncoding(db->handle(), "UTF8");
        }
        return db;
    }

protected:
    std::string m_host;
    unsigned short m_port;
    std::string m_database;
    std::string m_user;
    std::string m_password;
};

}

}

#endif //_QTL_MYSQL_POOL_H_

 include/qtl_sqlite.hpp

@@ -574,6 +574,9 @@
    void swap( blobbuf& other )
    {
        std::swap(m_blob, other.m_blob);
        std::swap(m_inbuf, other.m_inbuf);
        std::swap(m_outbuf, other.m_outbuf);
        std::swap(m_size, other.m_size);
        std::swap(m_inpos, other.m_inpos);
        std::swap(m_outpos, other.m_outpos);

@@ -649,7 +652,7 @@
        return this;
    }

    std::streamoff blob_size() const { return std::streamoff(m_size); }
    std::streamoff size() const { return std::streamoff(m_size); }

    void flush() 
    {
@@ -783,7 +786,7 @@
            if(m_outpos>=m_size)
                return traits_type::eof();
            if(sqlite3_blob_write(m_blob, &c, 1, m_outpos)!=SQLITE_OK)
                traits_type::eof();
                return traits_type::eof();
            auto intersection = interval_intersection(m_inpos, egptr()-eback(), m_outpos, 1);
            if(intersection.first!=intersection.second)
            {

 test/TestPostgres.cpp

@@ -40,13 +40,15 @@
{
    this->id = 0;
    TEST_ADD(TestPostgres::test_dual)
        TEST_ADD(TestPostgres::test_clear)
        TEST_ADD(TestPostgres::test_insert)
        TEST_ADD(TestPostgres::test_select)
        TEST_ADD(TestPostgres::test_update)
        TEST_ADD(TestPostgres::test_insert2)
        TEST_ADD(TestPostgres::test_iterator)
        TEST_ADD(TestPostgres::test_any)
    TEST_ADD(TestPostgres::test_clear)
    TEST_ADD(TestPostgres::test_insert)
    TEST_ADD(TestPostgres::test_select)
    TEST_ADD(TestPostgres::test_update)
    TEST_ADD(TestPostgres::test_insert2)
    TEST_ADD(TestPostgres::test_iterator)
    TEST_ADD(TestPostgres::test_insert_blob)
    TEST_ADD(TestPostgres::test_select_blob)
    TEST_ADD(TestPostgres::test_any)
}

inline void TestPostgres::connect(qtl::postgres::database& db)
@@ -188,6 +190,89 @@
    }
}

void hex_string(char* dest, const unsigned char* bytes, size_t count)
{
    for (size_t i = 0; i != count; i++)
    {
        sprintf(&dest[i * 2], "%02X", bytes[i] & 0xFF);
    }
}

void TestPostgres::test_insert_blob()
{
    qtl::postgres::database db;
    connect(db);

    try
    {
#ifdef _WIN32
        const char filename[] = "C:\\windows\\explorer.exe";
#else
        const char filename[] = "/bin/sh";
#endif //_WIN32
        qtl::postgres::transaction trans(db);
        qtl::postgres::large_object content = qtl::postgres::large_object::load(db.handle(), filename);
        TEST_ASSERT_MSG(content.oid()>0, "Cannot open test file.");
        fstream fs(filename, ios::binary | ios::in);
        unsigned char md5[16] = { 0 };
        char md5_hex[33] = { 0 };
        get_md5(fs, md5);
        hex_string(md5_hex, md5, sizeof(md5));
        printf("MD5 of file %s: %s.\n", filename, md5_hex);

        db.simple_execute("DELETE FROM test_blob");

        fs.clear();
        fs.seekg(0, ios::beg);
        db.query_first("INSERT INTO test_blob (filename, content, md5) values($1, $2, $3) returning id",
            forward_as_tuple(filename, content, qtl::const_blob_data(md5, sizeof(md5))), id);
        content.close();
        trans.commit();
    }
    catch (qtl::postgres::error& e)
    {
        ASSERT_EXCEPTION(e);
    }
}

void TestPostgres::test_select_blob()
{
    qtl::postgres::database db;
    connect(db);

    try
    {
        const char dest_file[] = "explorer.exe";

        unsigned char md5[16] = { 0 };
        std::string source_file;
        qtl::postgres::transaction trans(db);
        qtl::postgres::large_object content;
        qtl::const_blob_data md5_value;
        db.query_first("SELECT filename, content, md5 FROM test_blob WHERE id=$1", make_tuple(id),
            forward_as_tuple(source_file, content, qtl::blob_data(md5, sizeof(md5))));
        if (content.is_open())
        {
            content.save(dest_file);
            ifstream fs(dest_file, ios::binary | ios::in);
            TEST_ASSERT_MSG(fs, "Cannot open test file.");
            char md5_hex[33] = { 0 };
            hex_string(md5_hex, md5, sizeof(md5));
            printf("MD5 of file %s stored in database: %s.\n", source_file.data(), md5_hex);
            fs.clear();
            fs.seekg(0, ios::beg);
            get_md5(fs, md5);
            hex_string(md5_hex, md5, sizeof(md5));
            printf("MD5 of file %s: %s.\n", dest_file, md5_hex);
            content.close();
        }
    }
    catch (qtl::postgres::error& e)
    {
        ASSERT_EXCEPTION(e);
    }
}

void TestPostgres::test_any()
{
#ifdef _QTL_ENABLE_CPP17
@@ -218,6 +303,19 @@
#endif
}

void TestPostgres::get_md5(std::istream& is, unsigned char* result)
{
    std::array<char, 64 * 1024> buffer;
    MD5_CTX context;
    MD5Init(&context);
    while (!is.eof())
    {
        is.read(&buffer.front(), buffer.size());
        MD5Update(&context, (unsigned char*)buffer.data(), (unsigned int)is.gcount());
    }
    MD5Final(result, &context);
}

int main(int argc, char* argv[])
{
    Test::TextOutput output(Test::TextOutput::Verbose);

 test/TestPostgres.h

@@ -24,6 +24,8 @@
    void test_update();
    void test_insert2();
    void test_iterator();
    void test_insert_blob();
    void test_select_blob();
    void test_any();

private:

 test/vs/test_postgres/test_postgres.vcproj

@@ -177,6 +177,10 @@
            UniqueIdentifier="{4FC737F1-C7A5-4376-A066-2A32D752A2FF}"
            >
            <File
                RelativePath="..\..\md5.c"
                >
            </File>
            <File
                RelativePath="..\..\stdafx.cpp"
                >
            </File>
@@ -191,6 +195,10 @@
            UniqueIdentifier="{93995380-89BD-4b04-88EB-625FBE52EBFB}"
            >
            <File
                RelativePath="..\..\md5.h"
                >
            </File>
            <File
                RelativePath="..\..\stdafx.h"
                >
            </File>