PostgreSQL数据库扩展包——新增方法

野性酷女 2022-10-09 12:28 341阅读 0赞

## 1. 新增方法 ##

在/contrib文件夹下新增一个扩展包需要新增如下文件：

1.  扩展SQL文件（extension\_name.sql）：要执行的sql语句，比如函数声明，验证功能点等等。
2.  扩展控制文件（extension\_name.control）：主要是控制版本所用。
3.  扩展库文件（extension\_name.so，一般由.c 文件和makefile文件组成build生成。补充也可以是由.y 和.l 文件生成.c）：实现功能点的函数。

### 1.1扩展控制文件 ###

# test extension
     
    comment = 'only test publication'
    default_version = '1.0'
    module_pathname = '$libdir/test'
    relocatable = true

### 1.2 扩展SQL文件 ###

/* contrib/test/test--1.0.sql */
     
    --complain if script is sourced in psql rather than via ALTER EXTENSION
    CREATE TABLE test_table(oid integer,namespace_oid integer,name text,time timestamp);        /* 创建一个表格 */
     
     
    CREATE FUNCTION test_add_fun(integer,integer)           /* 创建一个函数 */
    RETURNS integer
    AS 'MODULE_PATHNAME' , 'test_add_fun'
    LANGUAGE C STRICT PARALLEL RESTRICTED;

### 1.3扩展库文件（.c） ###

#include "postgres.h"
    #include "fmgr.h"
     
    #ifdef PG_MODULE_MAGIC
    PG_MODULE_MAGIC;
    #endif
     
     
    PG_FUNCTION_INFO_V1(test_add_fun);
     
    Datum test_add_fun(PG_FUNCTION_ARGS);
     
    Datum test_add_fun(PG_FUNCTION_ARGS)
    { 
    	int sum ,a,b;
    	
    	a=PG_GETARG_INT32(0);
    	b=PG_GETARG_INT32(1);
     
    	sum = a + b;
    	PG_RETURN_INT32(sum);
    }

完成文件编写后，make & make install，然后通过 create extension test，加载插件。通过\\dx来查看插件是否安装成功。通过drop extension test来卸载插件。

## 2. 源码原理 ##

Create extension原理：程序先会解析控制文件，取出需要执行的sql版本号，同时创建extension oid并加入到pg\_extension中。然后解析相对应的sql文件。解析后执行其中的语句。如果有创建函数的话则会load .so文件(win32下.dll文件)。

Drop extension原理：将extension oid 从pg\_extension 删除掉并级联删除掉相关信息。

### 2.1 扩展控制文件解析 ###

控制文件的结构体如下：

typedef struct ExtensionControlFile
    { 
    	char	   *name;			/* name of the extension */
    	char	   *directory;		/* directory for script files */
    	char	   *default_version;	/* default install target version, if any */
    	char	   *module_pathname;	/* string to substitute for * MODULE_PATHNAME */
    	char	   *comment;		/* comment, if any */
    	char	   *schema;			/* target schema (allowed if !relocatable) */
    	bool		relocatable;	/* is ALTER EXTENSION SET SCHEMA supported? */
    	bool		superuser;		/* must be superuser to install? */
    	int		encoding;		/* encoding of the script file, or -1 */
    	List	   *requires;		/* names of prerequisite extensions */
    } ExtensionControlFile;

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70]

1.  parse\_extension\_control\_file()：解析控制文件，取出调用具体版本，文件名等信息。
2.  InsertExtensionTuple():把控制文件的信息保存到Tuple中。
3.  execute\_sql\_string():解析sql文件。先调用pg\_parse\_query将sql转换成parsetree\_list。
4.  然后执行pg\_analyze\_and\_rewrite(),pg\_plan\_queries()，ExecutorRun()对每一个parse tree重写执行。

### 2.2 扩展SQL文件解析 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70 1]

1.  get\_ext\_ver\_list()：遍历所有.sql文件获取所有版本节点。
2.  find\_install\_path()：根据目标节点选择最短路径以及相对应的起始节点。
3.  find\_update\_path()：最短路径算法Dijkstra的实现。  
    get\_extension\_script\_filename获取sql文件名，规则如下：
4.  版本命名的规则:扩展名 +“–”+“版本号”+ .sql (eg:cube–1.2.sql)
5.  版本与版本之间的升级采用“–”连接。(eg:cube–1.2–1.3.sql)
6.  Target 版本由控制文件中的default\_version确定。
7.  根据目标版本选择最短路径。

static ExtensionVersionInfo *
    find_install_path(List *evi_list, ExtensionVersionInfo *evi_target,
    				  List **best_path)
    { 
    	ExtensionVersionInfo *evi_start = NULL;
    	ListCell   *lc;
     
    	*best_path = NIL;
     
    	/* * We don't expect to be called for an installable target, but if we are, * the answer is easy: just start from there, with an empty update path. */
    	if (evi_target->installable)
    		return evi_target;
     
    	/* Consider all installable versions as start points */
    	foreach(lc, evi_list)
    	{ 
    		ExtensionVersionInfo *evi1 = (ExtensionVersionInfo *) lfirst(lc);
    		List	   *path;
     
    		if (!evi1->installable)
    			continue;
     
    		/* * Find shortest path from evi1 to evi_target; but no need to consider * paths going through other installable versions. */
                      //根据Dijkstra算法找到evi1 到evi_target的最短路径。
    		path = find_update_path(evi_list, evi1, evi_target, true, true); 
            
    		if (path == NIL)
    			continue;
     
    		/* Remember best path */
                     //遍历每个节点找到最短路径的起始节点，并保存最短路径。
    		if (evi_start == NULL ||
    			list_length(path) < list_length(*best_path) ||
    			(list_length(path) == list_length(*best_path) &&
    			 strcmp(evi_start->name, evi1->name) < 0))
    		{ 
    			evi_start = evi1;
    			*best_path = path;
    		}
    	}
     
    	return evi_start;
    }

### 2.3 扩展库文件解析 ###

![在这里插入图片描述][2021062422041664.png]

typedef struct FmgrInfo
    { 
    	PGFunction	fn_addr;		/* pointer to function or handler to be called */
    	Oid			fn_oid;			/* OID of function (NOT of handler, if any) */
    	short		fn_nargs;		/* number of input args (0..FUNC_MAX_ARGS) */
    	bool		fn_strict;		/* function is "strict" (NULL in => NULL out) */
    	bool		fn_retset;		/* function returns a set */
    	unsigned char fn_stats;		/* collect stats if track_functions > this */
    	void	   *fn_extra;		/* extra space for use by handler */
    	MemoryContext fn_mcxt;		/* memory context to store fn_extra in */
    	fmNodePtr	fn_expr;		/* expression parse tree for call, or NULL */
    } FmgrInfo;

1.  ProcedureCreate()：在pg\_proc.c中，主要是创建函数和存储过程。
2.  OidFunctionCall()：调用fmgr提供的接口。
3.  Internal\_load\_library()：先判断是否load，如果没有则通过调用dlopen()函数进行load，然后dlsym()查找相关函数。
4.  Dlopen()：在Unix下调用shl\_load()加载.so文件；在windows下调用LoadLibrary()加载.dll文件。
5.  Dlsym()：获取函数Magic\_func()和\_pg\_init()函数的地址，如果没获取到Magic\_func()则会报error。 对于\_PG\_init()可有可无。在Unix：调用shl\_findSym()；windows：GetProcAddress()获取函数地址。
6.  Fetch\_finfo\_record()：获取指定函数的地址，判断函数是否存在，不存在会报error。

static void *
    internal_load_library(const char *libname)
    { 
    	DynamicFileList *file_scanner;
    	PGModuleMagicFunction magic_func;
    	char	   *load_error;
    	struct stat stat_buf;
    	PG_init_t	PG_init;
     
    	  // 遍历file_scanner，判断文件是否load；
    	for (file_scanner = file_list;
    		 file_scanner != NULL &&
    		 strcmp(libname, file_scanner->filename) != 0;
    		 file_scanner = file_scanner->next)
    		;
     
    	if (file_scanner == NULL)
    	{ 
    		/* * Check for same files - different paths (ie, symlink or link) */
    		if (stat(libname, &stat_buf) == -1)
    			ereport(ERROR,
    					(errcode_for_file_access(),
    					 errmsg("could not access file \"%s\": %m",
    							libname)));
     
    		for (file_scanner = file_list;
    			 file_scanner != NULL &&
    			 !SAME_INODE(stat_buf, *file_scanner);
    			 file_scanner = file_scanner->next)
    			;
    	}
     
    	if (file_scanner == NULL)
    	{ 
    		/* * File not loaded yet. */
    		file_scanner = (DynamicFileList *)
    			malloc(offsetof(DynamicFileList, filename) + strlen(libname) + 1);
    		if (file_scanner == NULL)
    			ereport(ERROR,
    					(errcode(ERRCODE_OUT_OF_MEMORY),
    					 errmsg("out of memory")));
     
    		MemSet(file_scanner, 0, offsetof(DynamicFileList, filename));
    		strcpy(file_scanner->filename, libname);
    		file_scanner->device = stat_buf.st_dev;
    #ifndef WIN32
    		file_scanner->inode = stat_buf.st_ino;
    #endif
    		file_scanner->next = NULL;
     
    		file_scanner->handle = dlopen(file_scanner->filename, RTLD_NOW | RTLD_GLOBAL);
    		if (file_scanner->handle == NULL)
    		{ 
    			load_error = dlerror();
    			free((char *) file_scanner);
    			/* errcode_for_file_access might not be appropriate here? */
    			ereport(ERROR,
    					(errcode_for_file_access(),
    					 errmsg("could not load library \"%s\": %s",
    							libname, load_error)));
    		}
     
    		// 查找PG_MODULE_MAGIC的模块。实际上是个函数
    		magic_func = (PGModuleMagicFunction)
    			dlsym(file_scanner->handle, PG_MAGIC_FUNCTION_NAME_STRING);
    		if (magic_func)
    		{ 
    			const Pg_magic_struct *magic_data_ptr = (*magic_func) ();
     
    			if (magic_data_ptr->len != magic_data.len ||
    				memcmp(magic_data_ptr, &magic_data, magic_data.len) != 0)
    			{ 
    				/* copy data block before unlinking library */
    				Pg_magic_struct module_magic_data = *magic_data_ptr;
     
    				/* try to close library */
    				dlclose(file_scanner->handle);
    				free((char *) file_scanner);
     
    				/* issue suitable complaint */
    				incompatible_module_error(libname, &module_magic_data);
    			}
    		}
    		else
    		{ 
    			/* try to close library */
    			dlclose(file_scanner->handle);
    			free((char *) file_scanner);
    			/* complain */
    			ereport(ERROR,
    					(errmsg("incompatible library \"%s\": missing magic block",
    							libname),
    					 errhint("Extension libraries are required to use the PG_MODULE_MAGIC macro.")));
    		}
     
    		//查找_pg_init()函数，如果有则优先执行。
    		PG_init = (PG_init_t) dlsym(file_scanner->handle, "_PG_init");
    		if (PG_init)
    			(*PG_init) ();
     
    		/* OK to link it into list */
    		if (file_list == NULL)
    			file_list = file_scanner;
    		else
    			file_tail->next = file_scanner;
    		file_tail = file_scanner;
    	}
     
    	return file_scanner->handle;
    }

### 2.4 调用扩展包函数原理 ###

所有的函数调用都会调用OidInputFunctionCall()。

Datum
    OidInputFunctionCall(Oid functionId, char *str, Oid typioparam, int32 typmod)
    { 
    	FmgrInfo	flinfo;
     
    	fmgr_info(functionId, &flinfo);      
    	return InputFunctionCall(&flinfo, str, typioparam, typmod);
    }

Fmgr\_info()获取函数指针以及其他相关信息，对于系统函数则从Tuple中获取。对于扩展包中函数，则调用fmgr\_info\_C\_lang()来获取函数指针。

static void
    fmgr_info_C_lang(Oid functionId, FmgrInfo *finfo, HeapTuple procedureTuple)
    { 
    	
             ...
    		/* Look up the function itself */
    		user_fn = load_external_function(probinstring, prosrcstring, true,
    										 &libraryhandle);
     
    		/* Get the function information record (real or default) */
    		inforec = fetch_finfo_record(libraryhandle, prosrcstring);
     
    		/* Cache the addresses for later calls */
    		record_C_func(procedureTuple, user_fn, inforec);
     
    		pfree(prosrcstring);
    		pfree(probinstring);
     
             switch (inforec->api_version)
    	   { 
    		case 1:
    			/* New style: call directly */
    			finfo->fn_addr = user_fn;
    			break;
    		default:
    			/* Shouldn't get here if fetch_finfo_record did its job */
    			elog(ERROR, "unrecognized function API version: %d",
    				 inforec->api_version);
    			break;
    	  }
    }

最后InputFunctionCall(&flinfo, str, typioparam, typmod)中调用相关函数。

Datum
    InputFunctionCall(FmgrInfo *flinfo, char *str, Oid typioparam, int32 typmod)
    { 
         ...
    //这是个宏
    //#define FunctionCallInvoke(fcinfo) ((* (fcinfo)->flinfo->fn_addr) (fcinfo))
    	result = FunctionCallInvoke(fcinfo); 
     
    	/* Should get null result if and only if str is NULL */
    	if (str == NULL)
    	{ 
    		if (!fcinfo->isnull)
    			elog(ERROR, "input function %u returned non-NULL",
    				 flinfo->fn_oid);
    	}
    	else
    	{ 
    		if (fcinfo->isnull)
    			elog(ERROR, "input function %u returned NULL",
    				 flinfo->fn_oid);
    	}
     
    	return result;
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70]: /images/20221005/cedcbf9952d341ddab629433ccdaebc3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70 1]: /images/20221005/7a0468614952496da0c673d605c98a4b.png
[2021062422041664.png]: /images/20221005/5a15d8c567744523875413787a2159f4.png