PostgreSQL dblink使用过程

安装：

进入/root/postgresql-11.2/contrib/dblink

make && make install

切换到postgres用户

[root@fce40690-0e46-4603-e80e-ca351bda31ec dblink]# su - postgres

上一次登录：三 5月  8 03:47:54 UTC 2019pts/1 上

[postgres@fce40690-0e46-4603-e80e-ca351bda31ec ~]$ psql

psql (11.2)

Type "help" for help.

postgres=# create extension dblink;

ERROR:  extension "dblink" already exists

使用：

postgres=# select dblink_connect('mydb', 'dbname=mydb host=localhost');

 dblink_connect

----------------

 OK

(1 row)

postgres=# select * from dblink('mydb', 'select * from test') as test(id integer, info varchar(8));

ERROR:  value too long for type character varying(8)

postgres=# select * from dblink('mydb', 'select * from test') as test(id integer, info varchar(32));

 id |               info

----+----------------------------------

  1 | c53bc2b29be0ef76d863a53efed13ea1

  2 | 28fd974058eb778168482942c8894004

  3 | 6e90dbdf605adffe70761aa8521ab626

  4 | c32edb06e0f7e0067322f346ec60e663

  5 | 2eb0c9fcb931ede7440244745c686177

  6 | d34a41454758897e9b9d0c2dd1a972ae

  7 | 09d8e45db5f32a20e4e77ed51cf34b1a

  8 | 86cb5c36edda6fffd9f9c75aa9c3f0fe

  9 | 7ae28a3b304291dbcb96d19b92b42012

 10 | 581fddcce1c9e66a42ae4fad8a604e0d

(10 rows)

postgres=# select dblink_get_connections();

 dblink_get_connections

------------------------

 {mydb}

(1 row)

创建视图，来固化查询

postgres=# create view test_mydb as select * from dblink('mydb', 'select * from test') as test(id integer, info varchar(32));

CREATE VIEW

postgres=# select dblink_connect('mydb', 'dbname=mydb host=localhost');

 dblink_connect

----------------

 OK

(1 row)

postgres=# select * from test_mydb;

 id |               info

----+----------------------------------

  1 | c53bc2b29be0ef76d863a53efed13ea1

  2 | 28fd974058eb778168482942c8894004

  3 | 6e90dbdf605adffe70761aa8521ab626

  4 | c32edb06e0f7e0067322f346ec60e663

  5 | 2eb0c9fcb931ede7440244745c686177

  6 | d34a41454758897e9b9d0c2dd1a972ae

  7 | 09d8e45db5f32a20e4e77ed51cf34b1a

  8 | 86cb5c36edda6fffd9f9c75aa9c3f0fe

  9 | 7ae28a3b304291dbcb96d19b92b42012

 10 | 581fddcce1c9e66a42ae4fad8a604e0d

(10 rows)

执行没有返回结果的SQL

postgres=# select dblink_exec('mydb', 'insert into test values(1, ''dblink insert i am'')');

 dblink_exec

-------------

 INSERT 0 1

(1 row)

通fetch的方式分页获取数据：

postgres=# select dblink_open('mydb', 'foo', 'select * from test');

 dblink_open

-------------

 OK

(1 row)

                                                     ^

postgres=# select * from dblink_fetch('mydb', 'foo', 3) as test(id int, info text);

 id |               info

----+----------------------------------

  1 | c53bc2b29be0ef76d863a53efed13ea1

  2 | 28fd974058eb778168482942c8894004

  3 | 6e90dbdf605adffe70761aa8521ab626

(3 rows)

postgres=# select * from dblink_fetch('mydb', 'foo', 3) as (id int, info text);

 id |               info

----+----------------------------------

  4 | c32edb06e0f7e0067322f346ec60e663

  5 | 2eb0c9fcb931ede7440244745c686177

  6 | d34a41454758897e9b9d0c2dd1a972ae

(3 rows)

postgres=# (id int, info text);^C

postgres=# select * from dblink_fetch('mydb', 'foo', 3) as (id int, info text);

 id |               info

----+----------------------------------

  7 | 09d8e45db5f32a20e4e77ed51cf34b1a

  8 | 86cb5c36edda6fffd9f9c75aa9c3f0fe

  9 | 7ae28a3b304291dbcb96d19b92b42012

(3 rows)

postgres=# select * from dblink_fetch('mydb', 'foo', 3) as (id int, info text);

 id |               info

----+----------------------------------

 10 | 581fddcce1c9e66a42ae4fad8a604e0d

  1 | dblink insert i am

(2 rows)

使用异步调用：dblink_send_query、dblink_is_busy、dblink_get_notify和dblink_get_result

postgres=# select * from dblink_send_query('mydb', 'select * from test;');

 dblink_send_query

-------------------

                 1

(1 row)

postgres=# select * from dblink_get_result('mydb') as (id int, info text);

 id |               info

----+----------------------------------

  1 | c53bc2b29be0ef76d863a53efed13ea1

  2 | 28fd974058eb778168482942c8894004

  3 | 6e90dbdf605adffe70761aa8521ab626

  4 | c32edb06e0f7e0067322f346ec60e663

  5 | 2eb0c9fcb931ede7440244745c686177

  6 | d34a41454758897e9b9d0c2dd1a972ae

  7 | 09d8e45db5f32a20e4e77ed51cf34b1a

  8 | 86cb5c36edda6fffd9f9c75aa9c3f0fe

  9 | 7ae28a3b304291dbcb96d19b92b42012

 10 | 581fddcce1c9e66a42ae4fad8a604e0d

  1 | dblink insert i am

(11 rows)

postgres=# select * from dblink_send_query('mydb', 'select * from test;');

NOTICE:  could not send query: another command is already in progress

 dblink_send_query

-------------------

                 0

(1 row)

—获取到了最后一条，再执行新的SQL才会成功

postgres=# select * from dblink_get_result('mydb') as (id int, info text);

 id | info

----+------

(0 rows)

postgres=# select * from dblink_send_query('mydb', 'select * from test;');

 dblink_send_query

-------------------

                 1

(1 row)

获取错误信息：

postgres=# select dblink_error_message('mydb');

 dblink_error_message

----------------------

 OK

(1 row)

能否用dblink去连接oracle数据库呢？像oracle的dblink一样，连接SQL Server、MySQL、PostgreSQL？

答案是不行的，源码里面可以看到PostgreSQL的dblink是使用的pg的c语言接口去创建连接的，而不是使用ODBC来创建：

dblink_connect(PG_FUNCTION_ARGS)
{
	char	   *conname_or_str = NULL;
	char	   *connstr = NULL;
	char	   *connname = NULL;
	char	   *msg;
	PGconn	   *conn = NULL;
	remoteConn *rconn = NULL;

	dblink_init();

	if (PG_NARGS() == 2)
	{
		conname_or_str = text_to_cstring(PG_GETARG_TEXT_PP(1));
		connname = text_to_cstring(PG_GETARG_TEXT_PP(0));
	}
	else if (PG_NARGS() == 1)
		conname_or_str = text_to_cstring(PG_GETARG_TEXT_PP(0));

	if (connname)
		rconn = (remoteConn *) MemoryContextAlloc(TopMemoryContext,
												  sizeof(remoteConn));

	/* first check for valid foreign data server */
	connstr = get_connect_string(conname_or_str);
	if (connstr == NULL)
		connstr = conname_or_str;

	/* check password in connection string if not superuser */
	dblink_connstr_check(connstr);
	conn = PQconnectdb(connstr);

	if (PQstatus(conn) == CONNECTION_BAD)
	{
		msg = pchomp(PQerrorMessage(conn));
		PQfinish(conn);
		if (rconn)
			pfree(rconn);

		ereport(ERROR,
				(errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
				 errmsg("could not establish connection"),
				 errdetail_internal("%s", msg)));
	}

	/* check password actually used if not superuser */
	dblink_security_check(conn, rconn);

	/* attempt to set client encoding to match server encoding, if needed */
	if (PQclientEncoding(conn) != GetDatabaseEncoding())
		PQsetClientEncoding(conn, GetDatabaseEncodingName());

	if (connname)
	{
		rconn->conn = conn;
		createNewConnection(connname, rconn);
	}
	else
	{
		if (pconn->conn)
			PQfinish(pconn->conn);
		pconn->conn = conn;
	}

	PG_RETURN_TEXT_P(cstring_to_text("OK"));
}