Greenplum自定义外部表协议开发

ゝ一纸荒年。 2022-05-30 11:50 339阅读 0赞

**概述**

我们知道Greenplum支持gpfdist、http等多种形式的外部表协议。但如果gpfdist、http、file等协议都无法满足需求，Greenplum还支持自定义协议的外部表，用户可以自己定义一种协议访问自己的外部数据，且数据格式可以是标准的TEXT或CSV，也可以是自定义的数据格式。

自定义协议在一些特殊场景下比较有用，例如需要在自己的系统和Greenplum之间直接导入、导出数据。下面简单介绍一下开发一个自定义协议的步骤：

# 主要步骤 #

1、按照Greenplum预定义的API格式，实现几个C接口(import、export、validate\_urls)。这些接口将被编译成so的导出函数，并最终注册到GP（需要用户自己定义GP函数进行注册，后面会讲到。）

这几个接口的细节稍后再介绍，我们主要是在这些接口中，通过Greenplum提供的API，实现从Greenplum获取数据，或者向Greenplum导入数据。

2、拷贝上述so文件到gp安装目录：greenplum-db/lib/postgresql/，然后定义一系列数据库函数对so中的接口进行封装：

CREATE FUNCTION myread() RETURNS integer
    as '$libdir/gpextprotocol.so', 'demoprot_import'
    LANGUAGE C STABLE;
    
    CREATE FUNCTION mywrite() RETURNS integer
    as '$libdir/gpextprotocol.so', 'demoprot_export'
    LANGUAGE C STABLE;
    
    #可选函数
    CREATE OR REPLACE FUNCTION myvalidate() RETURNS void
    AS '$libdir/gpextprotocol.so', 'demoprot_validate_urls'
    LANGUAGE C STABLE;

3、基于上面定义这些数据库函数，定义自定义协议myprot。 其中myvalidatorfunc 是可选的。

CREATE TRUSTED PROTOCOL myprot(
    writefunc='mywrite',
    readfunc='myread',
    validatorfunc='myvalidate');

4. 授权用户使用上述协议

GRANT ALL ON PROTOCOL myprot TO otheruser

5、基于上述协议，创建可读或可写的外部表。

PS：自定义协议的外部表对异常数据处理和标准协议外部表相同。

CREATE WRITABLE EXTERNAL TABLE ext_sales(LIKE sales)
    LOCATION ('myprot://<meta>/<meta>/…')
    FORMAT 'TEXT';
    CREATE READABLE EXTERNAL TABLE ext_sales(LIKE sales)
    LOCATION('myprot://<meta>/<meta>/…')
    FORMAT 'TEXT';

# 自定义协议核心代码 #

## 导出函数 ##

实现将数据从Greenplum导出到外部表的过程。

Greenplum 外部表调用框架代码会根据sql语句，将查询到的数据组织成指定格式（如TEXT或CSV）放到一个buffer中，用户调用API获取该buffer的地址，从中读取数据，并按照自己的业务逻辑将其发送到自定义外部表。

其处理流程大概如下：

初始化外部连接->向外部系统发数据->关闭外部系统连接。

export函数会被GP外部表框架多次调用（取决于数据量的多少），需要我们结合GP提供的API自己判断连接是否初始化(初始化成功后将连接信息保存到自定义数据区)、数据是否发送完成。

### 初始化连接 ###

当export函数被调用时，首先调用 EXTPROTOCOL\_GET\_USER\_CTX 获取自定义信息(可用于保存外部表连接的描述信息)，如果为NULL，则需要执行初始化连接的动作，并调用 EXTPROTOCOL\_SET\_USER\_CTX 保存上述信息。

### **发送数据** ###

Datum  
    myprot_export(PG_FUNCTION_ARGS) 
    {
    ……
      /* ======================================== 
       *      DO THE EXPORT 
       * ======================================== */ 
      /* 当我们执行 insert into ext_test_table(xx,xx) values(xxxxxxxx)或insert into ext_test_table select * from internal_test_table 等sql语句时， 
       * gp会自动把数据组织成CSV或TEXT等格式放入buf中，我们只需要将其发送给外部系统就可以了。当然根据实际业务逻辑的需要，可能要对数据格式进行转换处理。 */
      data   = EXTPROTOCOL_GET_DATABUF(fcinfo); 
      datlen   = EXTPROTOCOL_GET_DATALEN(fcinfo); 
      
      if(datlen > 0) 
      {  
        /* 将数据写入(发送到)外部表。本Demo直接写入指定的本地文件中 */
        wrote = fwrite(data, 1, datlen, myData->file); 
        
        if (ferror(myData->file)) 
          ereport(ERROR, 
            (errcode_for_file_access(), 
             errmsg("myprot_import: could not read from file
                \"%s\": %m", 
                myData->filename))); 
      } 
      PG_RETURN_INT32((int)wrote);  
    }

### 关闭连接 ###

通过 EXTPROTOCOL\_IS\_LAST\_CALL 判断是否还有剩余数据。如没有其他数据，说明导出已经完成(静态数据的情况下)。需要关闭自定义外部表连接。

## 导入函数 ##

和导出函数类似，也需要完成初始化连接->接收数据->关闭连接的过程。

### 接收数据 ###

Datum myprot_import(PG_FUNCTION_ARGS)  
    {
    ……
     /* ========================================== 
       *          DO THE IMPORT 
       * ========================================== */ 
      data    = EXTPROTOCOL_GET_DATABUF(fcinfo); 
      datlen  = EXTPROTOCOL_GET_DATALEN(fcinfo); 
      
      /* read some bytes (with fread in this example, but normally
         in some other method over the network) */
      if(datlen > 0) 
      { 
    // GP 的API会帮我们申请好buf内存空间，我们按照自己的业务逻辑从外部表系统获取数据，并组织成外部表定义时约定的格式如CSV或TEXT，然后放入buf中，返回实际读到的长度就可以了
        nread = fread(data, 1, datlen, myData->file); 
        if (ferror(myData->file)) 
          ereport(ERROR, 
            (errcode_for_file_access(), 
              errmsg("myprot_import: could not write to file
                \"%s\": %m", 
                myData->filename))); 
      }
      PG_RETURN_INT32((int)nread); // 通常如果导入函数返回0，Greenplum框架认为导入完成。接下来调用EXTPROTOCOL_IS_LAST_CALL会返回true
    }

### 关闭连接 ###

数据导入Greenplum会面临一个问题，如何判断导入完成。因为外部数据源中的数据是静态的（比如已经存在于文本文件），还是动态的（比如来自某个消息队列或脚本程序动态生成），这个是不确定的。所以需要开发者根据自己的业务逻辑进行处理。

通常如果导入函数返回0，Greenplum框架认为导入完成。接下来调用EXTPROTOCOL\_IS\_LAST\_CALL会返回TRUE。也可以调用 EXTPROTOCOL\_SET\_LAST\_CALL 强制设置为TRUE。

EXTPROTOCOL_SET_LAST_CALL(fcinfo);

## (可选)validator函数 ##

比如检查LOCATION中URL的数量等

## 自定义外部协议API介绍 ##

GP提供了一系列API用于在自定义协议中控制数据交互：

/* ---- 读/写函数API ------*/
    CALLED_AS_EXTPROTOCOL(fcinfo)         /* 判断当前函数是否被外部表管理器调用  */
    EXTPROTOCOL_GET_URL(fcinfo)(fcinfo)   /* 获取外部表数据URL,初始化时需要调用该API  */
    EXTPROTOCOL_GET_DATABUF(fcinfo)       /* 获取和GP交互的缓冲区地址: 
    									   * 从GP导出数据时，GP会把数据填充到这个缓冲区，然后我们将其发送到自己的系统；
    									   * 向GP导入数据时，我们要自己控制从外部系统获取的数据，填入该缓冲区。
               							   * 缓冲区中的数据格式应该和定义外部表时指定的格式相同，如CSV或TEXT
    EXTPROTOCOL_GET_DATALEN(fcinfo)       /* 上述数据的长度 */
    EXTPROTOCOL_GET_SCANQUALS(fcinfo) 
    EXTPROTOCOL_GET_USER_CTX(fcinfo)      /* 获取用户自定义的上下文信息 */
    EXTPROTOCOL_IS_LAST_CALL(fcinfo)      /* 判断是否为最后一次调用(数据导入/导出完成)：
    									   * 从GP往外导时，源数据往往是静态的，GP自己知道什么时候结束，会设置相关信息。我们在调用该API得知导出完成后，执行一些清理关闭的动作；
                                           * 从外部系统向GP导入时，必须正确设置自定义read接口的返回值。当read接口返回0时，GP会认为导入已经完成。此后调用EXTPROTOCOL_IS_LAST_CALL会返回true。  */
    EXTPROTOCOL_SET_LAST_CALL(fcinfo)     /* 可以通过这个API自己设置什么时候结束  */
    EXTPROTOCOL_SET_USER_CTX(fcinfo, p)   /* 设置用户自定义上下文信息,和 GET_USER_CTX对应。一般可用该字段存放全局的上下文信息，比如，和外部系统连接相关的信息。
                                           * 在Demo中，如果该信息为NULL，认为是初次调用，执行与外部系统连接初始化动作，然后将连接上下文set进去，后续直接通过get使用。  */
    
    /* ------ 验证器函数API ------*/
    CALLED_AS_EXTPROTOCOL_VALIDATOR(fcinfo)        /* 一般在validate_urls中调用，判断前述函数是否被外部表管理器调用  */
    EXTPROTOCOL_VALIDATOR_GET_URL_LIST(fcinfo)     /* 获取外部表LOCATION URL列表 */
    EXTPROTOCOL_VALIDATOR_GET_NUM_URLS(fcinfo)     /* 获取url数量 */
    EXTPROTOCOL_VALIDATOR_GET_NTH_URL(fcinfo, n)   /* 获取第n个url */

## **编译命令** ##

GP源码 contrib\\extprotocol 目录给出了Makefile 文件，但真正开发自定义协议时难免会用到自己的头文件、lib库，需要修改Makefile。Demo对应的编译命令(gpadmin系统账户登陆执行）

cc -fpic -c gpextprotocol.c -I$GPHOME/include/postgresql/server 
    cc -shared -o gpextprotocol.so gpextprotocol.o

例如，笔者需要将MPP和kafka继承，编译中需要添加librdkafka的头文件和lib，命令如下（静态链接kafka库）

cc -fpic -c gpextprotocol.c -I$GPHOME/include/postgresql/server -I/usr/local/include/librdkafka/
    cc -shared -o gpextprotocol.so gpextprotocol.o librdkafka.a -lsasl2

PS：

GP官方文档中有很详细的介绍和Demo，本文根据官方文档和实际开发经历整理。

以下是GP 5.2版本源码中提供的Demo

#include "postgres.h"
    #include "fmgr.h"
    #include "funcapi.h"
    
    #include "access/extprotocol.h"
    #include "catalog/pg_proc.h"
    #include "utils/array.h"
    #include "utils/builtins.h"
    #include "utils/memutils.h"
    
    #include "catalog/pg_exttable.h"
    
    
    typedef struct DemoUri
    {
    	char	   *protocol;
    	char	   *path;
    	
    }	DemoUri;
    
    static DemoUri *ParseDemoUri(const char *uri_str);
    static void FreeDemoUri(DemoUri* uri);
    
    
    /* Do the module magic dance */
    PG_MODULE_MAGIC;
    PG_FUNCTION_INFO_V1(demoprot_export);
    PG_FUNCTION_INFO_V1(demoprot_import);
    PG_FUNCTION_INFO_V1(demoprot_validate_urls);
    
    Datum demoprot_export(PG_FUNCTION_ARGS);
    Datum demoprot_import(PG_FUNCTION_ARGS);
    Datum demoprot_validate_urls(PG_FUNCTION_ARGS);
    
    
    typedef struct {
    	char	  *url;
    	char	  *filename;
    	FILE	  *file;
    } extprotocol_t;
    
    static void check_ext_options(const FunctionCallInfo fcinfo)
    {
            ListCell *cell;
            Relation rel = EXTPROTOCOL_GET_RELATION(fcinfo);
            ExtTableEntry *exttbl = GetExtTableEntry(rel->rd_id);
            List *options = exttbl->options;
    
            foreach(cell, options) {
                    DefElem *def = (DefElem *) lfirst(cell);
                    char *key = def->defname;
    
                    if (key && strcasestr(key, "database") && !strcasestr(key, "greenplum")) {
                            ereport(ERROR, (0, errmsg("This is greenplum.")));
                    }
            }
    }
    
    /*
     * Import data into GPDB.
     */
    Datum 
    demoprot_import(PG_FUNCTION_ARGS)
    {
    	extprotocol_t   *myData;
    	char			*data;
    	int				 datlen;
    	size_t			 nread = 0;
    
    	/* Must be called via the external table format manager */
    	if (!CALLED_AS_EXTPROTOCOL(fcinfo))
    		elog(ERROR, "extprotocol_import: not called by external protocol manager");
    
    	/* Get our internal description of the protocol */
    	myData = (extprotocol_t *) EXTPROTOCOL_GET_USER_CTX(fcinfo);
    
    	if(EXTPROTOCOL_IS_LAST_CALL(fcinfo))
    	{
    		/* we're done receiving data. close our connection */
    		if(myData && myData->file)
    			if(fclose(myData->file))
    				ereport(ERROR,
    						(errcode_for_file_access(),
    						 errmsg("could not close file \"%s\": %m",
    								 myData->filename)));
    		
    		PG_RETURN_INT32(0);
    	}	
    
    	if (myData == NULL)
    	{
    		/* first call. do any desired init */
    		
    		const char	*p_name = "demoprot";
    		DemoUri		*parsed_url;
    		char		*url = EXTPROTOCOL_GET_URL(fcinfo);
    			
    		myData 			 = palloc(sizeof(extprotocol_t));
    				
    		myData->url 	 = pstrdup(url);
    		parsed_url 		 = ParseDemoUri(myData->url);
    		myData->filename = pstrdup(parsed_url->path);
    	
    		if(strcasecmp(parsed_url->protocol, p_name) != 0)
    			elog(ERROR, "internal error: demoprot called with a different protocol (%s)",
    						parsed_url->protocol);
    
    		/* An example of checking options */
    		check_ext_options(fcinfo);
    
    		FreeDemoUri(parsed_url);
    		
    		/* open the destination file (or connect to remote server in other cases) */
    		myData->file = fopen(myData->filename, "r");
    		
    		if (myData->file == NULL)
    			ereport(ERROR,
    					(errcode_for_file_access(),
    					 errmsg("demoprot_import: could not open file \"%s\" for reading: %m",
    							myData->filename)));
    		
    		EXTPROTOCOL_SET_USER_CTX(fcinfo, myData);
    	}
    
    	/* =======================================================================
    	 *                            DO THE IMPORT
    	 * ======================================================================= */
    	
    	data 	= EXTPROTOCOL_GET_DATABUF(fcinfo);
    	datlen 	= EXTPROTOCOL_GET_DATALEN(fcinfo);
    
    	if(datlen > 0)
    	{
    		nread = fread(data, 1, datlen, myData->file);
    		if (ferror(myData->file))
    			ereport(ERROR,
    					(errcode_for_file_access(),
    					 errmsg("demoprot_import: could not write to file \"%s\": %m",
    							 myData->filename)));		
    	}
    
    	
    	PG_RETURN_INT32((int)nread);
    }
    
    /*
     * Export data out of GPDB.
     */
    Datum 
    demoprot_export(PG_FUNCTION_ARGS)
    {
    	extprotocol_t   *myData;
    	char			*data;
    	int				 datlen;
    	size_t			 wrote = 0;
    
    	/* Must be called via the external table format manager */
    	if (!CALLED_AS_EXTPROTOCOL(fcinfo))
    		elog(ERROR, "extprotocol_export: not called by external protocol manager");
    
    	/* Get our internal description of the protocol */
    	myData = (extprotocol_t *) EXTPROTOCOL_GET_USER_CTX(fcinfo);
    
    	if(EXTPROTOCOL_IS_LAST_CALL(fcinfo))
    	{
    		/* we're done sending data. close our connection */
    		if(myData && myData->file)
    			if(fclose(myData->file))
    				ereport(ERROR,
    						(errcode_for_file_access(),
    						 errmsg("could not close file \"%s\": %m",
    								 myData->filename)));
    		
    		PG_RETURN_INT32(0);
    	}	
    
    	if (myData == NULL)
    	{
    		/* first call. do any desired init */
    		
    		const char	*p_name = "demoprot";
    		DemoUri		*parsed_url;
    		char		*url = EXTPROTOCOL_GET_URL(fcinfo);
    			
    		myData 			 = palloc(sizeof(extprotocol_t));
    				
    		myData->url 	 = pstrdup(url);
    		parsed_url 		 = ParseDemoUri(myData->url);
    		myData->filename = pstrdup(parsed_url->path);
    	
    		if(strcasecmp(parsed_url->protocol, p_name) != 0)
    			elog(ERROR, "internal error: demoprot called with a different protocol (%s)",
    						parsed_url->protocol);
    
    		FreeDemoUri(parsed_url);
    		
    		/* open the destination file (or connect to remote server in other cases) */
    		myData->file = fopen(myData->filename, "a");
    		
    		if (myData->file == NULL)
    			ereport(ERROR,
    					(errcode_for_file_access(),
    					 errmsg("demoprot_export: could not open file \"%s\" for writing: %m",
    							myData->filename)));
    		
    		EXTPROTOCOL_SET_USER_CTX(fcinfo, myData);
    	}
    
    	/* =======================================================================
    	 *                            DO THE EXPORT
    	 * ======================================================================= */
    	
    	data 	= EXTPROTOCOL_GET_DATABUF(fcinfo);
    	datlen 	= EXTPROTOCOL_GET_DATALEN(fcinfo);
    
    	if(datlen > 0)
    	{
    		wrote = fwrite(data, 1, datlen, myData->file);
    		if (ferror(myData->file))
    			ereport(ERROR,
    					(errcode_for_file_access(),
    					 errmsg("demoprot_import: could not read from file \"%s\": %m",
    							 myData->filename)));		
    	}
    
    	PG_RETURN_INT32((int)wrote);
    }
    
    Datum 
    demoprot_validate_urls(PG_FUNCTION_ARGS)
    {
    	int					nurls;
    	int					i;
    	ValidatorDirection	direction;
    	
    	/* Must be called via the external table format manager */
    	if (!CALLED_AS_EXTPROTOCOL_VALIDATOR(fcinfo))
    		elog(ERROR, "demoprot_validate_urls: not called by external protocol manager");
    
    	nurls 		= EXTPROTOCOL_VALIDATOR_GET_NUM_URLS(fcinfo);
    	direction 	= EXTPROTOCOL_VALIDATOR_GET_DIRECTION(fcinfo);
    	
    	/*
    	 * Dumb example 1: search each url for a substring 
    	 * we don't want to be used in a url. in this example
    	 * it's 'secured_directory'.
    	 */
    	for (i = 1 ; i <= nurls ; i++)
    	{
    		char *url = EXTPROTOCOL_VALIDATOR_GET_NTH_URL(fcinfo, i);
    		
    		if (strstr(url, "secured_directory") != 0)
    		{
                ereport(ERROR,
                        (errcode(ERRCODE_PROTOCOL_VIOLATION),
                         errmsg("using 'secured_directory' in a url isn't allowed ")));
    		}
    	}
    	
    	/*
    	 * Dumb example 2: set a limit on the number of urls 
    	 * used. In this example we limit readable external
    	 * tables that use our protocol to 2 urls max.
    	 */
    	if(direction == EXT_VALIDATE_READ && nurls > 2)
    	{
            ereport(ERROR,
                    (errcode(ERRCODE_PROTOCOL_VIOLATION),
                     errmsg("more than 2 urls aren't allowed in this protocol ")));
    	}
    	
    	PG_RETURN_VOID();
    }
    
    /* --- utility functions --- */
    
    static 
    DemoUri *ParseDemoUri(const char *uri_str)
    {
    	DemoUri	   *uri = (DemoUri *) palloc0(sizeof(DemoUri));
    	int			protocol_len;
    
     	uri->path = NULL;
     	uri->protocol = NULL;
     	
    	/*
    	 * parse protocol
    	 */
    	char *post_protocol = strstr(uri_str, "://");
    		
    	if(!post_protocol)
    	{
    		ereport(ERROR,
    				(errcode(ERRCODE_SYNTAX_ERROR),
    				 errmsg("invalid demo prot URI \'%s\'", uri_str)));
    	}
    
    	protocol_len = post_protocol - uri_str;
    	uri->protocol = (char *) palloc0 (protocol_len + 1);
    	strncpy(uri->protocol, uri_str, protocol_len);				
    
    	/* make sure there is more to the uri string */
    	if (strlen(uri_str) <= protocol_len)
    		ereport(ERROR,
    				(errcode(ERRCODE_SYNTAX_ERROR),
    		errmsg("invalid demo prot URI \'%s\' : missing path", uri_str)));
    
    	/*
    	 * parse path
    	 */
    	uri->path = pstrdup(uri_str + protocol_len + strlen("://"));
    	
    	return uri;
    }
    
    static
    void FreeDemoUri(DemoUri *uri)
    {
    	if (uri->path)
    		pfree(uri->path);
    	if (uri->protocol)
    		pfree(uri->protocol);
    	
    	pfree(uri);
    }