• 关于使用mysql_fdw无法查询长度超过64K字段的原因分析

    LightDB 23.3 版本修复无法使用mysql_fdw查询超过64K字段的问题。

    注：笔者使用的是mysql_fdw REL-2_7_0，这个bug已在REL-2_8_0中修复，我们来复盘一下造成这个问题的原因所在。

    问题复现

    1. -- mysql上创建测试表
    2. create table fdwtest(a longtext);
    3. insert into fdwtest select repeat('a',65534);
    4. insert into fdwtest select repeat('b',65535);
    5. insert into fdwtest select repeat('c',65536);
    6. insert into fdwtest select repeat('d',65537);
    7. insert into fdwtest select repeat('e',65538);
    8.  
    9. mysql> select substr(a,1,10),length(a) from fdwtest;
    10. +----------------+-----------+
    11. | substr(a,1,10) | length(a) |
    12. +----------------+-----------+
    13. | aaaaaaaaaa     |     65534 |
    14. | bbbbbbbbbb     |     65535 |
    15. | cccccccccc     |     65536 |
    16. | dddddddddd     |     65537 |
    17. | eeeeeeeeee     |     65538 |
    18. +----------------+-----------+
    19. 5 rows in set (0.00 sec)
    1. -- 在pg中使用mysql_fdw查询mysql表
    2. -- load extension first time after install
    3. CREATE EXTENSION mysql_fdw;
    4.  
    5. -- create server object
    6. CREATE SERVER mysql_server
    7. 	FOREIGN DATA WRAPPER mysql_fdw
    8. 	OPTIONS(host '127.0.0.1', port '3306');
    9.  
    10. -- create user mapping
    11. CREATE USER MAPPING FOR lightdb
    12. 	SERVER mysql_server
    13. 	OPTIONS (username 'ludeng', password 'ludeng');
    14.  
    15. -- create foreign table
    16. CREATE FOREIGN TABLE fdwtest
    17. (
    18. 	a text
    19. )
    20. SERVER mysql_server
    21. OPTIONS (dbname 'ludeng', table_name 'fdwtest');
    22.  
    23. -- 在msyql中的长度超过64k
    24. lightdb@postgres=# select substr(a,1,10),length(a) from fdwtest;
    25.    substr   | length 
    26. ------------+--------
    27.  aaaaaaaaaa |  65534
    28.  bbbbbbbbbb |  65535
    29.  cccccccccc |  65536
    30. (3 rows)

    可以看到，对于长度超过64K的字段，已经无法通过mysql_fdw来查询到了。

    mysql_fdw工作原理

    我们知道mysql数据库对外提供了一套c语言的API接口libmysqlclient，相关资料可以参考官网MySQL :: MySQL 8.0 C API Developer Guide。而mysq_fdw作为postgresql的一个插件，正是使用了libmysqlclient API来实现在postgresql上直接查询mysql数据库表的目的，使得postgresql用户可以像直接查询postgresql数据库一样查询mysql数据库，大致的执行流程如下

    libmysqlclient使用

    现在我们知道了mysql_fdw的大致工作原理是通过libmysqlclient与mysql去交互，那么我们就来简单使用一下libmysqlclient提供的c语言API，对libmysqlclient有个大致的了解。由于mysql_fdw的实现是基于C API Prepared Statement Interface的，那么我们就简单使用下相关的API，细节可参考官网

    在mysql上使用c语言prepared statement API的大致步骤如下
    1、使用mysql_stmt_init创建一个prepared statement处理器，调用mysql_stmt_prepare时传入一个字符串(sql语句)来创建一个prepared statement。当调用mysql_stmt_prepare时，mysql客户端/服务端协议执行了如下操作：
    1-服务端解析(包括词法和语法解析)传入的sql语句，并返回给客户端OK状态(返回值为0)，同时还返回一个statement ID(该statement ID是正整数，用于标识所有本次sql执行的各个步骤)，以及参数个数、列数和结果集元数据
    2-客户端使用statement ID来进行接下来的操作，以便服务器知道这是针对那个sql语句的


    2、通过mysql_stmt_bind_param来设置参数，所有参数必须均被设置，否则执行时会报错或返回不可预期的结果。


    3、调用mysql_stmt_execute()执行sql语句，具体步骤为
    1-客户端发送sql语句的参数给服务器，即绑定参数步骤
    2-服务端用客户端提供的statment ID找到是哪个sql语句，用客户端提供的参数替换占位符并执行sql语句。若sql语句需要返回结果集，服务端返回结果数据给客户端，否则只需要返回OK状态，以及几行数据被更新/删除/插入


    4、如果sql语句是一个select语句或者其它返回结果的语句，那么可以调用mysql_stmt_result_metadata获取结果集的元数据。结果集元数据是个MYSQL_RES结构，而结果集本身也是MYSQL_RES结构，结果集元数据中包含了查询的列数以及每列的相关信息。


    5、若一个sql语句会产生结果集，需要通过调用mysql_stmt_bind_result来绑定数据缓冲区(data buffer)，而数据缓冲区将用于后续获取数据。


    6、调用mysql_stmt_fetch将数据逐行获取到缓冲区，直到所有数据行均被获取到。


    7、如有必要，重复步骤3至6。通过调用mysql_stmt_bind_param更改prepared statement的参数值，可以重复mysql_stmt_execute来重新执行该语句。


    8、当sql语句执行结束时，调用mysql_stmt_close关闭处理器(statement handler)，这将释放所有资源，且handler此时已处于无效状态。


    9、若使用mysql_stmt_result_metadata获取了结果集的元数据，需调用mysql_free_result来释放元数据相关资源防止内存泄露。

    在了解了API的大致使用之后，我们直接上一个hello world例子：

    1. #include 
    2. #include "mysql/mysql.h"
    3. #include 
    4. #include 
    5.  
    6. #define STRING_SIZE 50
    7. #define SELECT_SAMPLE "select length(a),substr(a,1,10) from fdwtest limit 5"
    8. #define COLUMN_NUMS	2
    9. typedef char bool;
    10.  
    11. int main()
    12. {
    13. 	MYSQL *mysql; // connection
    14. 	MYSQL_STMT *stmt;
    15. 	MYSQL_BIND bind[COLUMN_NUMS];
    16. 	MYSQL_RES *prepare_meta_result;
    17. 	MYSQL_TIME ts;
    18. 	unsigned long length[COLUMN_NUMS];
    19. 	int param_count, column_count, row_count;
    20. 	short small_data;
    21. 	int int_data;
    22. 	char str_data[STRING_SIZE];
    23. 	bool is_null[COLUMN_NUMS];
    24. 	bool error[COLUMN_NUMS];
    25. 	
    26. 	// create connection to MySQL 
    27. 	mysql = mysql_init(NULL);
    28. 	if(mysql_real_connect(mysql, "127.0.0.1", "ludeng", "ludeng", "ludeng", 3306, NULL, 0) == NULL) { 
    29. 		fprintf(stderr, "sorry, no database connection ...\n");
    30. 		return 1;
    31. 	}
    32. 	
    33. 	/* Prepare a SELECT query to fetch data from test_table */
    34. 	stmt = mysql_stmt_init(mysql);
    35. 	if (!stmt)
    36. 	{
    37. 		fprintf(stderr, " mysql_stmt_init(), out of memory\n");
    38. 		exit(0);
    39. 	}
    40. 	if (mysql_stmt_prepare(stmt, SELECT_SAMPLE, strlen(SELECT_SAMPLE)))
    41. 	{
    42. 		fprintf(stderr, " mysql_stmt_prepare(), SELECT failed\n");
    43. 		fprintf(stderr, " %s\n", mysql_stmt_error(stmt));
    44. 		exit(0);
    45. 	}
    46. 	fprintf(stdout, " prepare, SELECT successful\n");
    47. 	
    48. 	/* Get the parameter count from the statement */
    49. 	param_count= mysql_stmt_param_count(stmt);
    50. 	fprintf(stdout, " total parameters in SELECT: %d\n", param_count);
    51. 	if (param_count != 0) /* validate parameter count */
    52. 	{
    53. 		fprintf(stderr, " invalid parameter count returned by MySQL\n");
    54. 		exit(0);
    55. 	}
    56. 	
    57. 	/* Execute the SELECT query */
    58. 	if (mysql_stmt_execute(stmt))
    59. 	{
    60. 		fprintf(stderr, " mysql_stmt_execute(), failed\n");
    61. 		fprintf(stderr, " %s\n", mysql_stmt_error(stmt));
    62. 		exit(0);
    63. 	}
    64. 	
    65. 	/* Fetch result set meta information */
    66. 	prepare_meta_result = mysql_stmt_result_metadata(stmt);
    67. 	if (!prepare_meta_result)
    68. 	{
    69. 		fprintf(stderr," mysql_stmt_result_metadata(), returned no meta information\n");
    70. 		fprintf(stderr, " %s\n", mysql_stmt_error(stmt));
    71. 		exit(0);
    72. 	}
    73. 	
    74. 	/* Get total columns in the query */
    75. 	column_count= mysql_num_fields(prepare_meta_result);
    76. 	fprintf(stdout," total columns in SELECT statement: %d\n", column_count);
    77. 	if (column_count != 2) /* validate column count */
    78. 	{
    79. 		fprintf(stderr, " invalid column count returned by MySQL\n");
    80. 		exit(0);
    81. 	}
    82. 	/* Bind the result buffers for all 2 columns before fetching them */
    83. 	memset(bind, 0, sizeof(bind));
    84. 	/* INTEGER COLUMN */
    85. 	bind[0].buffer_type= MYSQL_TYPE_LONG;
    86. 	bind[0].buffer= (char *)&int_data;
    87. 	bind[0].is_null= &is_null[0];
    88. 	bind[0].length= &length[0];
    89. 	bind[0].error= &error[0];
    90. 	/* STRING COLUMN */
    91. 	bind[1].buffer_type= MYSQL_TYPE_STRING;
    92. 	bind[1].buffer= (char *)str_data;
    93. 	bind[1].buffer_length= STRING_SIZE;
    94. 	bind[1].is_null= &is_null[1];
    95. 	bind[1].length= &length[1];
    96. 	bind[1].error= &error[1];
    97. 	
    98. 	/* Bind the result buffers */
    99. 	if (mysql_stmt_bind_result(stmt, bind))
    100. 	{
    101. 		fprintf(stderr, " mysql_stmt_bind_result() failed\n");
    102. 		fprintf(stderr, " %s\n", mysql_stmt_error(stmt));
    103. 		exit(0);
    104. 	}
    105. 	
    106. 	/* Now buffer all results to client (optional step) */
    107. 	if (mysql_stmt_store_result(stmt))
    108. 	{
    109. 		fprintf(stderr, " mysql_stmt_store_result() failed\n");
    110. 		fprintf(stderr, " %s\n", mysql_stmt_error(stmt));
    111. 		exit(0);
    112. 	}
    113. 	
    114. 	/* Fetch all rows */
    115. 	row_count= 0;
    116. 	fprintf(stdout, "Fetching results ...\n");
    117. 	while (!mysql_stmt_fetch(stmt))
    118. 	{
    119. 		row_count++;
    120. 		fprintf(stdout, " row %d\n", row_count);
    121. 		/* column 1 */
    122. 		fprintf(stdout, " column1 (integer) : ");
    123. 		if (is_null[0])
    124. 			fprintf(stdout, " NULL\n");
    125. 		else
    126. 			fprintf(stdout, " %d(%ld)\n", int_data, length[0]);
    127. 			
    128. 		/* column 2 */
    129. 		fprintf(stdout, " column2 (string) : ");
    130. 		if (is_null[1])
    131. 			fprintf(stdout, " NULL\n");
    132. 		else
    133. 			fprintf(stdout, " %s(%ld)\n", str_data, length[1]);
    134. 	}
    135. 	
    136. 	/* Validate rows fetched */
    137. 	fprintf(stdout, " total rows fetched: %d\n", row_count);
    138. 	if (row_count != 5)
    139. 	{
    140. 		fprintf(stderr, " MySQL failed to return all rows\n");
    141. 		exit(0);
    142. 	}
    143. 	
    144. 	/* Free the prepared result metadata */
    145. 	mysql_free_result(prepare_meta_result);
    146. 	/* Close the statement */
    147. 	if (mysql_stmt_close(stmt))
    148. 	{
    149. 		/* mysql_stmt_close() invalidates stmt, so call */
    150. 		/* mysql_error(mysql) rather than mysql_stmt_error(stmt) */
    151. 		fprintf(stderr, " failed while closing the statement\n");
    152. 		fprintf(stderr, " %s\n", mysql_error(mysql));
    153. 		exit(0);
    154. 	}
    155. 	
    156. 	return 0;
    157. }
    158.  
    159. //cc -o testLibMysqlClientStmt testLibMysqlClientStmt.c -I/usr/include/mysql/ -L /usr/lib64/mysql/ -lmysqlclient -g
    160.  
    161.  
    162.  
    163. [lightdb@node101 libmysqlclient]$ ./testLibMysqlClientStmt 
    164.  prepare, SELECT successful
    165.  total parameters in SELECT: 0
    166.  total columns in SELECT statement: 2
    167. Fetching results ...
    168.  row 1
    169.  column1 (integer) :  65534(4)
    170.  column2 (string) :  aaaaaaaaaa(10)
    171.  row 2
    172.  column1 (integer) :  65535(4)
    173.  column2 (string) :  bbbbbbbbbb(10)
    174.  row 3
    175.  column1 (integer) :  65536(4)
    176.  column2 (string) :  cccccccccc(10)
    177.  row 4
    178.  column1 (integer) :  65537(4)
    179.  column2 (string) :  dddddddddd(10)
    180.  row 5
    181.  column1 (integer) :  65538(4)
    182.  column2 (string) :  eeeeeeeeee(10)
    183.  total rows fetched: 5

    mysql_fdw源码分析

    注：这是2-8-0版本的mysql_fdw源码

    在mysql_fdw源码中，大致可以认为在mysqlBeginForeignScan中连接mysql数据库，而在mysqlIterateForeignScan中就是从mysql数据库获取查询数据，调用栈如下

    mysqlBeginForeignScan            
        mysql_get_connection        
        mysql_stmt_init                
        mysql_stmt_prepare            
        mysql_stmt_result_metadata    
        mysql_fetch_fields            
        若查询表存在类型为text的字段, 设置festate->has_var_size_col = true #本次新增#
        mysql_bind_result            
        mysql_stmt_attr_set    设置stmt属性 #本次新增#            
        mysql_stmt_bind_result        

    mysqlIterateForeignScan        
        ExecClearTuple                
        bind_stmt_params_and_exec    
            mysql_stmt_execute        
            mysql_stmt_store_result    在客户端结果集中缓存查询返回的全量结果，这里将field->max_length置为63358 #本次新增#
            mysql_bind_result(loop)    对TEXT类型重新绑定对应的内存 #本次新增#
            mysql_stmt_bind_result    为结果集中的输出字段建立缓冲(数据和长度的缓冲) #本次新增#
        mysql_stmt_fetch             
        mysql_convert_to_pg            
        heap_form_tuple                
        ExecStoreHeapTuple            
        内存释放和返回TupleTableSlot

    1. 所以造成数据查询不到的本质就在于：
    2. rc = mysql_stmt_fetch(festate->stmt);   //返回101(MYSQL_DATA_TRUNCATED)，导致查询结果不再返回给客户端了 
    3. 	if (rc == 0)
    4. 	{
    5. 		foreach(lc, festate->retrieved_attrs)
    6. 		{
    7. 			int			attnum = lfirst_int(lc) - 1;
    8. 			Oid			pgtype = TupleDescAttr(attinmeta->tupdesc, attnum)->atttypid;
    9. 			int32		pgtypmod = TupleDescAttr(attinmeta->tupdesc, attnum)->atttypmod;
    10.  
    11. 			nulls[attnum] = festate->table->column[attid].is_null;
    12. 			if (!festate->table->column[attid].is_null)
    13. 				dvalues[attnum] = mysql_convert_to_pg(pgtype, pgtypmod,
    14. 													  &festate->table->column[attid]);
    15.  
    16. 			attid++;
    17. 		}
    18.  
    19. 		ExecClearTuple(tupleSlot);
    20.  
    21. 		if (list_length(fdw_private) >= mysqlFdwPrivateScanTList)
    22. 		{
    23. 			/* Construct tuple with whole-row references. */
    24. 			tup = mysql_get_tuple_with_whole_row(festate, dvalues, nulls);
    25. 		}
    26. 		else
    27. 		{
    28. 			/* Form the Tuple using Datums */
    29. 			tup = heap_form_tuple(attinmeta->tupdesc, dvalues, nulls);
    30. 		}
    31.  
    32. ...
    33. }

    查看libmysqlclient官方对于mysql_stmt_attr_set的注解：If STMT_ATTR_UPDATE_MAX_LENGTH set to 1, causes mysql_stmt_store_result() to update the metadata MYSQL_FIELD->max_length value。调用mysql_stmt_attr_set将属性STMT_ATTR_UPDATE_MAX_LENGTH设置为true后，使得调用mysql_stmt_store_result后会更新结果集中字段的长度。对于我们的案例，长度为65537的字段来说，经过上述2个步骤后，会将MYSQL_FIELD->max_length value的值设置为65537，这时我们根据MYSQL_FIELD->max_length value重新申请长度为65537字节的内存，并再次将输出字段绑定到刚申请的内存即可，后续就能正常输出长度为65537的字段了。

    总结

    这个问题当然是mysql_fdw的bug，修改本身也比较简单。通过使用mysql_stmt_attr_set设置STMT_ATTR_UPDATE_MAX_LENGTH为1，调用mysql_stmt_store_result更新结果集中字段的长度即可。

    但是mysql_fdw的官方修复只是针对在pg上创建text类型的字段时的场景，并未包含对varchar和char字段的类似处理，也就是说若pg表中存在varchar和char字段，且varchar或char对应的mysql表的字段长度超过64KB，无法查询mysql_fdw表超长字段的问题依然存在。因为pg的字符类型字段最大可以支持10485760(the length must be greater than zero and cannot exceed 10485760)，远超64KB。

    实际的修复也很简单，当表中存在varchar和char字段时，只要保持跟text一样的处理方式即可，在lightdb中23.3版本，我们修复了这个问题。

    因此，若lightdb用户碰到mysql fdw表字段超过64K而无法正确查询时，只需将lightdb升级到23.3即可。

  • 相关阅读:
    Day17.1：静态与非静态的详解
    vue3项目实战中的接口调用方法（一）async/await用法 对axios二次封装 实现异步请求
    单机Linux下搭建MongoDB副本集-三节点
    canvas画布绘制线条样式：粗细，圆角，拐角等
    【微前端开发环境下，加载远程子应用的实战。】
    「网页开发｜前端开发｜Vue」10 vuex模块化：将数据划分成不同modules分别管理
    华为浏览器风险提示 - 解决方案
    【网络安全】使用meterpreter进行远控、Mysql注入、反弹型XSS攻防
    虾皮物流价格是多少？如何计算？
    java计算机毕业设计web家教管理系统源码+mysql数据库+系统+lw文档+部署
  • 原文地址：https://blog.csdn.net/u012796067/article/details/132617318