!2947 dblink支持通过odbc连接异构数据库

Merge pull request !2947 from 陈志凯/master

GNUmakefile.in

@@ -83,6 +83,7 @@ install:
 	$(MAKE) install_pldebugger
 	$(MAKE) -C contrib/postgres_fdw $@
 	$(MAKE) -C contrib/hstore $@
+	$(MAKE) -C contrib/dblink $@
 	$(MAKE) -C $(root_builddir)/privategauss/kernel/extension/packages $@
 	+@echo "openGauss installation complete."
 endif

build/script/aarch64_opengauss_list

@@ -83,6 +83,9 @@
 ./share/postgresql/extension/mot_fdw.control
 ./share/postgresql/extension/postgres_fdw--1.0.sql
 ./share/postgresql/extension/postgres_fdw.control
+./share/postgresql/extension/dblink--1.0.sql
+./share/postgresql/extension/dblink--unpackaged--1.0.sql
+./share/postgresql/extension/dblink.control
 ./share/postgresql/timezone/GB-Eire
 ./share/postgresql/timezone/Turkey
 ./share/postgresql/timezone/Kwajalein
@@ -757,6 +760,7 @@
 ./lib/postgresql/pg_upgrade_support.so
 ./lib/postgresql/java/pljava.jar
 ./lib/postgresql/postgres_fdw.so
+./lib/postgresql/dblink.so
 ./lib/postgresql/pgoutput.so
 ./lib/libpljava.so
 ./lib/libpq.a

build/script/opengauss_release_list_mini

@@ -79,6 +79,9 @@
 ./share/postgresql/extension/gsredistribute--unpackaged--1.0.sql
 ./share/postgresql/extension/postgres_fdw--1.0.sql
 ./share/postgresql/extension/postgres_fdw.control
+./share/postgresql/extension/dblink--1.0.sql
+./share/postgresql/extension/dblink--unpackaged--1.0.sql
+./share/postgresql/extension/dblink.control
 ./share/postgresql/timezone/GB-Eire
 ./share/postgresql/timezone/Turkey
 ./share/postgresql/timezone/Kwajalein
@@ -752,6 +755,7 @@
 ./lib/postgresql/pg_upgrade_support.so
 ./lib/postgresql/java/pljava.jar
 ./lib/postgresql/postgres_fdw.so
+./lib/postgresql/dblink.so
 ./lib/libpljava.so
 ./lib/libpq.a
 ./lib/libpq.so

build/script/opengauss_release_list_ubuntu_single

@@ -82,6 +82,9 @@
 ./share/postgresql/extension/gsredistribute--unpackaged--1.0.sql
 ./share/postgresql/extension/postgres_fdw--1.0.sql
 ./share/postgresql/extension/postgres_fdw.control
+./share/postgresql/extension/dblink--1.0.sql
+./share/postgresql/extension/dblink--unpackaged--1.0.sql
+./share/postgresql/extension/dblink.control
 ./share/postgresql/timezone/GB-Eire
 ./share/postgresql/timezone/Turkey
 ./share/postgresql/timezone/Kwajalein
@@ -755,6 +758,7 @@
 ./lib/postgresql/pg_upgrade_support.so
 ./lib/postgresql/java/pljava.jar
 ./lib/postgresql/postgres_fdw.so
+./lib/postgresql/dblink.so
 ./lib/libpljava.so
 ./lib/libpq.a
 ./lib/libpq.so

build/script/x86_64_opengauss_list

@@ -83,6 +83,9 @@
 ./share/postgresql/extension/mot_fdw.control
 ./share/postgresql/extension/postgres_fdw--1.0.sql
 ./share/postgresql/extension/postgres_fdw.control
+./share/postgresql/extension/dblink--1.0.sql
+./share/postgresql/extension/dblink--unpackaged--1.0.sql
+./share/postgresql/extension/dblink.control
 ./share/postgresql/timezone/GB-Eire
 ./share/postgresql/timezone/Turkey
 ./share/postgresql/timezone/Kwajalein
@@ -757,6 +760,7 @@
 ./lib/postgresql/pg_upgrade_support.so
 ./lib/postgresql/java/pljava.jar
 ./lib/postgresql/postgres_fdw.so
+./lib/postgresql/dblink.so
 ./lib/postgresql/pgoutput.so
 ./lib/libpljava.so
 ./lib/libpq.a

contrib/dblink/Makefile

@@ -3,7 +3,7 @@
 MODULE_big = dblink
 OBJS	= dblink.o
 PG_CPPFLAGS = -I$(libpq_srcdir)
-SHLIB_LINK = $(libpq)
+SHLIB_LINK = $(libpq) -lodbc
 SHLIB_PREREQS = submake-libpq
 
 EXTENSION = dblink

contrib/dblink/dblink--1.0.sql

@@ -221,3 +221,8 @@ CREATE FUNCTION dblink_get_notify(
 RETURNS setof record
 AS 'MODULE_PATHNAME', 'dblink_get_notify'
 LANGUAGE C STRICT;
+
+CREATE FUNCTION dblink_get_drivername()
+RETURNS text
+AS 'MODULE_PATHNAME', 'dblink_get_drivername'
+LANGUAGE C STRICT;

contrib/dblink/dblink--unpackaged--1.0.sql

@@ -44,3 +44,4 @@ ALTER EXTENSION dblink ADD function dblink_cancel_query(text);
 ALTER EXTENSION dblink ADD function dblink_error_message(text);
 ALTER EXTENSION dblink ADD function dblink_get_notify();
 ALTER EXTENSION dblink ADD function dblink_get_notify(text);
+ALTER EXTENSION dblink ADD function dblink_get_drivername();

contrib/dblink/dblink.h

@@ -35,6 +35,113 @@
 #define DBLINK_H
 
 #include "fmgr.h"
+#include "sql.h"
+#include "sqlext.h"
+
+
+typedef struct ODBCconn {
+    SQLHENV       envHandle;        /* Handle ODBC environment */
+    SQLHDBC       connHandle;       /* Handle connection */
+    SQLHSTMT      stmt;             /* Handle sql */
+} ODBCconn;
+
+typedef struct storeInfo {
+    FunctionCallInfo fcinfo;
+    Tuplestorestate* tuplestore;
+    AttInMetadata* attinmeta;
+    MemoryContext tmpcontext;
+    char** cstrs;
+    /* temp storage for results to avoid leaks on exception */
+    PGresult* last_res;
+    PGresult* cur_res;
+} storeInfo;
+
+typedef struct LinkInfo {
+    SQLCHAR* username;      /* odbc username */
+    SQLCHAR* password;      /* odbc password */
+    SQLCHAR* drivername;    /* odbc driver name */
+} LinkInfo;
+
+class Linker : public BaseObject {
+public:
+    virtual void finish() = 0;
+    virtual text* exec(char* conname, const char* sql, bool fail) = 0;
+    virtual char* errorMsg() = 0;
+    virtual int isBusy() = 0;
+    virtual char* cancel(PGcancel* cancel) = 0;
+    virtual int sendQuery(char* sql) = 0;
+    virtual char* open(char* conname, char* sql, bool fail) = 0;
+    virtual char* close(char* conname, char* sql, bool fail) = 0;
+    virtual void getResult(char* conname, FunctionCallInfo fcinfo, char* sql, bool fail) = 0;
+    virtual void queryResult(ReturnSetInfo* rsinfo, const char* conname, storeInfo* sinfo, const char* sql, bool fail) = 0;
+    virtual void fetch(char* conname, FunctionCallInfo fcinfo, const char* sql, bool fail, char* curname) = 0;
+    virtual void getNotify(ReturnSetInfo* rsinfo) = 0;
+};
+
+typedef struct remoteConn {
+    Linker* linker;
+} remoteConn;
+
+typedef struct dblink_session_context {
+    remoteConn* pconn;
+    HTAB* remoteConnHash;
+} dblink_session_context;
+
+/*
+ *	Following is list that holds multiple remote connections.
+ *	Calling convention of each dblink function changes to accept
+ *	connection name as the first parameter. The connection list is
+ *	much like ecpg e.g. a mapping between a name and a PGconn object.
+ */
+
+typedef struct remoteConnHashEnt {
+    char name[NAMEDATALEN];
+    remoteConn* rconn;
+} remoteConnHashEnt;
+
+class PQLinker : public Linker {
+public: 
+    PGconn* conn;
+    PGresult* res;
+    int openCursorCount;        /* The number of open cursors */
+    bool newXactForCursor;      /* Opened a transaction for a cursor */
+public:
+    PQLinker(char* connstr);
+    void finish();
+    text* exec(char* conname, const char* sql, bool fail);
+    char* errorMsg();
+    int isBusy();
+    char* cancel(PGcancel* cancel);
+    int sendQuery(char* sql);
+    char* open(char* conname, char* sql, bool fail);
+    char* close(char* conname, char* sql, bool fail);
+    void getResult(char* conname, FunctionCallInfo fcinfo, char* sql, bool fail);
+    void queryResult(ReturnSetInfo* rsinfo, const char* conname, storeInfo* sinfo, const char* sql, bool fail);
+    void fetch(char* conname, FunctionCallInfo fcinfo, const char* sql, bool fail, char* curname);
+    void getNotify(ReturnSetInfo* rsinfo);
+};
+
+
+class ODBCLinker : public Linker {
+public:
+    SQLHENV       envHandle;        /* Handle ODBC environment */
+    SQLHDBC       connHandle;       /* Handle connection */
+    SQLHSTMT      stmt;             /* Handle sql */
+public:
+    ODBCLinker(char* connstr_or_name);
+    void finish();
+    text* exec(char* conname, const char* sql, bool fail);
+    char* errorMsg();
+    int isBusy();
+    char* cancel(PGcancel* cancel);
+    int sendQuery(char* sql);
+    char* open(char* conname, char* sql, bool fail);
+    char* close(char* conname, char* sql, bool fail);
+    void getResult(char* conname, FunctionCallInfo fcinfo, char* sql, bool fail);
+    void queryResult(ReturnSetInfo* rsinfo, const char* conname, storeInfo* sinfo, const char* sql, bool fail);
+    void fetch(char* conname, FunctionCallInfo fcinfo, const char* sql, bool fail, char* curname);
+    void getNotify(ReturnSetInfo* rsinfo);
+};
 
 /*
  * External declarations
@@ -58,6 +165,7 @@ extern "C" Datum dblink_build_sql_delete(PG_FUNCTION_ARGS);
 extern "C" Datum dblink_build_sql_update(PG_FUNCTION_ARGS);
 extern "C" Datum dblink_current_query(PG_FUNCTION_ARGS);
 extern "C" Datum dblink_get_notify(PG_FUNCTION_ARGS);
+extern "C" Datum dblink_get_drivername(PG_FUNCTION_ARGS);
 extern "C" void set_extension_index(uint32 index);
 extern "C" void init_session_vars(void);
 

src/gausskernel/Makefile

@@ -622,6 +622,7 @@ endif
 	cp $(SECUREDYNAMICLIB_HOME)/libsecurec* '$(DESTDIR)$(libdir)/'
 ifneq (, $(findstring __USE_NUMA, $(CFLAGS)))
 	cp $(ZSTD_LIB_PATH)/libzstd* '$(DESTDIR)$(libdir)/'
+	cp $(LIBODBC_LIB_PATH)/libodbc* '$(DESTDIR)$(libdir)/'
 	cp $(NUMA_LIB_PATH)/* '$(DESTDIR)$(libdir)/'
 endif
 ifeq ($(enable_mot), yes)
@@ -685,7 +686,7 @@ ifeq ($(enable_lite_mode), no)
 endif
 endif
 	cp '$(ZSTD_LIB_PATH)'/libzstd.so* '$(DESTDIR)$(libdir)/'
-
+	cp '$(LIBODBC_LIB_PATH)'/libodbc* '$(DESTDIR)$(libdir)/'
 	cp -d '$(LIBCURL_LIB_PATH)'/libcurl.so* '$(DESTDIR)$(libdir)/'
 endif