Extending gpfdist in Cloudberry Database to Support

王平10304955 · 王平10304955 · commit c8be42336b20 · 2025-07-15T09:52:01.000+08:00
SFTP Protocol for Data Ingestion -- add LIBSSH2 macro

ADD LIBSSH2 macro Label the SFTP-related code to indicate its
characteristics.
diff --git a/src/backend/utils/misc/fstream/fstream.c b/src/backend/utils/misc/fstream/fstream.c
@@ -1011,7 +1011,7 @@ fstream_open(const char *path, const struct fstream_options *options,
 		struct gpfxdist_t* transform = (i == 0) ? options->transform : NULL;
 
 		gfile_close(&fs->fd);
-
+#ifdef LIBSSH2
 		if (is_sftp)
 		{
 			if (gfile_open_sftp(&fs->fd, fs->glob.gl_pathv[i], fs->glob.gl_username[i], fs->glob.gl_passwd[i],
@@ -1022,6 +1022,7 @@ fstream_open(const char *path, const struct fstream_options *options,
 				return 0;
 			}
 		}
+#endif
 		else
 		{
 			if (gfile_open(&fs->fd, fs->glob.gl_pathv[i], gfile_open_flags(options->forwrite, options->usesync),
@@ -1082,7 +1083,7 @@ static int nextFile(fstream_t*fs)
 	if (fs->fidx < fs->glob.gl_pathc)
 	{
 		fs->skip_header_line = fs->options.header;
-
+#ifdef LIBSSH2
 		if (fs->fd.is_sftp)
 		{
 			if (gfile_open_sftp(&fs->fd, fs->glob.gl_pathv[fs->fidx], fs->glob.gl_username[fs->fidx], fs->glob.gl_passwd[fs->fidx],
@@ -1094,7 +1095,7 @@ static int nextFile(fstream_t*fs)
 				return 1;
 			}
 		}
-
+#endif
 		else
 		{
 			if (gfile_open(&fs->fd, fs->glob.gl_pathv[fs->fidx], GFILE_OPEN_FOR_READ,
@@ -1105,7 +1106,7 @@ static int nextFile(fstream_t*fs)
 				fs->ferror = "unable to open file";
 				return 1;
 			}
-		}
+		}	
 	}
 
 	return 0;
diff --git a/src/backend/utils/misc/fstream/gfile.c b/src/backend/utils/misc/fstream/gfile.c
@@ -40,8 +40,11 @@
 #include <sys/file.h>   /* for flock */
 #include <unistd.h>
 
+#ifdef LIBSSH2
 #include <libssh2.h>
 #include <libssh2_sftp.h>
+#endif
+
 #include <arpa/inet.h>
 #include <net/if.h>
 
@@ -151,6 +154,7 @@ writewinpipe(gfile_t* fd, void* ptr, size_t size)
 	return i;
 }
 
+#ifdef LIBSSH2
 static ssize_t
 sftp_read(gfile_t *fd, void *ptr, size_t size)
 {
@@ -189,6 +193,7 @@ sftp_close(gfile_t *fd)
 	libssh2_exit();
 	return 0;
 }
+#endif
 
 #ifdef HAVE_LIBBZ2
 static void *
@@ -1378,11 +1383,12 @@ gfile_close(gfile_t*fd)
 			{
 				fd->close(fd);
 			}
-
+#ifdef LIBSSH2
 			if (fd->is_sftp)
 			{
 				sftp_close(fd);
 			}
+#endif
 			else
 			{
 				if (fd->is_win_pipe)
@@ -1410,6 +1416,7 @@ gfile_close(gfile_t*fd)
 	return ret;
 }
 
+#ifdef LIBSSH2
 int gfile_open_sftp(gfile_t *fd, const char *fpath, const char *sftp_uname, const char *sftp_passwd, const char *sftp_hostaddr,
 					const char *sftp_port, int flags, int *response_code, const char **response_string, struct gpfxdist_t *transform)
 {
@@ -1497,6 +1504,7 @@ int gfile_open_sftp(gfile_t *fd, const char *fpath, const char *sftp_uname, cons
 
 	return 1;
 }
+#endif
 
 ssize_t 
 gfile_read(gfile_t *fd, void *ptr, size_t len)
@@ -1548,7 +1556,7 @@ off_t gfile_get_compressed_position(gfile_t *fd)
 	return fd->compressed_position;
 }
 
-
+#ifdef LIBSSH2
 int sftp_open(gfile_t *fd, const char *fpath, const char *sftp_uname, const char *sftp_passwd, const char *sftp_hostaddr,
 			 const char *sftp_port)
 {
@@ -1651,3 +1659,4 @@ void sftp_free(gfile_t *fd)
 	fd->sock = -1;
 	libssh2_exit();
 }
+#endif
diff --git a/src/include/fstream/gfile.h b/src/include/fstream/gfile.h
@@ -12,6 +12,11 @@
 #include <zstd.h>
 #endif
 
+#ifdef LIBSSH2
+#include <libssh2.h>
+#include <libssh2_sftp.h>
+#endif
+
 #ifdef WIN32
 #include <windows.h>
 #endif
@@ -26,9 +31,6 @@ typedef _int64 ssize_t;
 
 #include "c.h"
 
-#include <libssh2.h>
-#include <libssh2_sftp.h>
-
 #ifdef WIN32
 typedef BOOL bool_t;
 #else
@@ -116,8 +118,6 @@ int gfile_open_flags(int writing, int usesync);
 #define GFILE_OPEN_FOR_WRITE_SYNC   2
 
 int gfile_open(gfile_t* fd, const char* fpath, int flags, int* response_code, const char** response_string, struct gpfxdist_t* transform);
-int gfile_open_sftp(gfile_t *fd, const char *fpath, const char *sftp_uname, const char *sftp_passwd, const char *sftp_hostaddr,
-					const char *sftp_port, int flags, int *response_code, const char **response_string, struct gpfxdist_t *transform);
 int gfile_close(gfile_t*fd);
 off_t gfile_get_compressed_size(gfile_t*fd);
 off_t gfile_get_compressed_position(gfile_t*fd);
@@ -127,7 +127,11 @@ void gfile_printf_then_putc_newline(const char*format,...) pg_attribute_printf(1
 void*gfile_malloc(size_t size);
 void gfile_free(void*a);
 
+#ifdef LIBSSH2
+int gfile_open_sftp(gfile_t *fd, const char *fpath, const char *sftp_uname, const char *sftp_passwd, const char *sftp_hostaddr,
+					const char *sftp_port, int flags, int *response_code, const char **response_string, struct gpfxdist_t *transform);
 extern int sftp_open(gfile_t *fd, const char *fpath, const char *sftp_uname, const char *sftp_passwd, const char *sftp_hostaddr,
 					const char *sftp_port);
 extern void sftp_free(gfile_t *fd);
 #endif
+#endif

Original file line number	Diff line number	Diff line change
`@@ -1011,7 +1011,7 @@ fstream_open(const char path, const struct fstream_options options,`
`1011`	`1011`	`struct gpfxdist_t* transform = (i == 0) ? options->transform : NULL;`
`1012`	`1012`
`1013`	`1013`	`gfile_close(&fs->fd);`
`1014`		`-`
	`1014`	`+#ifdef LIBSSH2`
`1015`	`1015`	`if (is_sftp)`
`1016`	`1016`	`{`
`1017`	`1017`	`if (gfile_open_sftp(&fs->fd, fs->glob.gl_pathv[i], fs->glob.gl_username[i], fs->glob.gl_passwd[i],`
`@@ -1022,6 +1022,7 @@ fstream_open(const char path, const struct fstream_options options,`
`1022`	`1022`	`return 0;`
`1023`	`1023`	`}`
`1024`	`1024`	`}`
	`1025`	`+#endif`
`1025`	`1026`	`else`
`1026`	`1027`	`{`
`1027`	`1028`	`if (gfile_open(&fs->fd, fs->glob.gl_pathv[i], gfile_open_flags(options->forwrite, options->usesync),`
`@@ -1082,7 +1083,7 @@ static int nextFile(fstream_t*fs)`
`1082`	`1083`	`if (fs->fidx < fs->glob.gl_pathc)`
`1083`	`1084`	`{`
`1084`	`1085`	`fs->skip_header_line = fs->options.header;`
`1085`		`-`
	`1086`	`+#ifdef LIBSSH2`
`1086`	`1087`	`if (fs->fd.is_sftp)`
`1087`	`1088`	`{`
`1088`	`1089`	`if (gfile_open_sftp(&fs->fd, fs->glob.gl_pathv[fs->fidx], fs->glob.gl_username[fs->fidx], fs->glob.gl_passwd[fs->fidx],`
`@@ -1094,7 +1095,7 @@ static int nextFile(fstream_t*fs)`
`1094`	`1095`	`return 1;`
`1095`	`1096`	`}`
`1096`	`1097`	`}`
`1097`		`-`
	`1098`	`+#endif`
`1098`	`1099`	`else`
`1099`	`1100`	`{`
`1100`	`1101`	`if (gfile_open(&fs->fd, fs->glob.gl_pathv[fs->fidx], GFILE_OPEN_FOR_READ,`
`@@ -1105,7 +1106,7 @@ static int nextFile(fstream_t*fs)`
`1105`	`1106`	`fs->ferror = "unable to open file";`
`1106`	`1107`	`return 1;`
`1107`	`1108`	`}`
`1108`		`- }`
	`1109`	`+ }`
`1109`	`1110`	`}`
`1110`	`1111`
`1111`	`1112`	`return 0;`
Original file line number	Diff line number	Diff line change
`@@ -40,8 +40,11 @@`
`40`	`40`	`#include <sys/file.h> /* for flock */`
`41`	`41`	`#include <unistd.h>`
`42`	`42`
	`43`	`+#ifdef LIBSSH2`
`43`	`44`	`#include <libssh2.h>`
`44`	`45`	`#include <libssh2_sftp.h>`
	`46`	`+#endif`
	`47`	`+`
`45`	`48`	`#include <arpa/inet.h>`
`46`	`49`	`#include <net/if.h>`
`47`	`50`
`@@ -151,6 +154,7 @@ writewinpipe(gfile_t* fd, void* ptr, size_t size)`
`151`	`154`	`return i;`
`152`	`155`	`}`
`153`	`156`
	`157`	`+#ifdef LIBSSH2`
`154`	`158`	`static ssize_t`
`155`	`159`	`sftp_read(gfile_t fd, void ptr, size_t size)`
`156`	`160`	`{`
`@@ -189,6 +193,7 @@ sftp_close(gfile_t *fd)`
`189`	`193`	`libssh2_exit();`
`190`	`194`	`return 0;`
`191`	`195`	`}`
	`196`	`+#endif`
`192`	`197`
`193`	`198`	`#ifdef HAVE_LIBBZ2`
`194`	`199`	`static void *`
`@@ -1378,11 +1383,12 @@ gfile_close(gfile_t*fd)`
`1378`	`1383`	`{`
`1379`	`1384`	`fd->close(fd);`
`1380`	`1385`	`}`
`1381`		`-`
	`1386`	`+#ifdef LIBSSH2`
`1382`	`1387`	`if (fd->is_sftp)`
`1383`	`1388`	`{`
`1384`	`1389`	`sftp_close(fd);`
`1385`	`1390`	`}`
	`1391`	`+#endif`
`1386`	`1392`	`else`
`1387`	`1393`	`{`
`1388`	`1394`	`if (fd->is_win_pipe)`
`@@ -1410,6 +1416,7 @@ gfile_close(gfile_t*fd)`
`1410`	`1416`	`return ret;`
`1411`	`1417`	`}`
`1412`	`1418`
	`1419`	`+#ifdef LIBSSH2`
`1413`	`1420`	`int gfile_open_sftp(gfile_t fd, const char fpath, const char sftp_uname, const char sftp_passwd, const char *sftp_hostaddr,`
`1414`	`1421`	`const char sftp_port, int flags, int response_code, const char *response_string, struct gpfxdist_t transform)`
`1415`	`1422`	`{`
`@@ -1497,6 +1504,7 @@ int gfile_open_sftp(gfile_t fd, const char fpath, const char *sftp_uname, cons`
`1497`	`1504`
`1498`	`1505`	`return 1;`
`1499`	`1506`	`}`
	`1507`	`+#endif`
`1500`	`1508`
`1501`	`1509`	`ssize_t`
`1502`	`1510`	`gfile_read(gfile_t fd, void ptr, size_t len)`
`@@ -1548,7 +1556,7 @@ off_t gfile_get_compressed_position(gfile_t *fd)`
`1548`	`1556`	`return fd->compressed_position;`
`1549`	`1557`	`}`
`1550`	`1558`
`1551`		`-`
	`1559`	`+#ifdef LIBSSH2`
`1552`	`1560`	`int sftp_open(gfile_t fd, const char fpath, const char sftp_uname, const char sftp_passwd, const char *sftp_hostaddr,`
`1553`	`1561`	`const char *sftp_port)`
`1554`	`1562`	`{`
`@@ -1651,3 +1659,4 @@ void sftp_free(gfile_t *fd)`
`1651`	`1659`	`fd->sock = -1;`
`1652`	`1660`	`libssh2_exit();`
`1653`	`1661`	`}`
	`1662`	`+#endif`