pdo_pgsql: unbuffered fetching

fetch part of the lazy fetch mode for Pdo\Pgsql
Starts fixing #15287

Author: outtersg

ext/pdo_pgsql/pgsql_statement.c

@@ -58,6 +58,17 @@
 
 
 
+static void pgsql_stmt_cancel(pdo_pgsql_stmt *S)
+{
+	if (S->is_running_unbuffered) {
+		PGcancel *cancel = PQgetCancel(S->H->server);
+		char errbuf[256];
+		PQcancel(cancel, errbuf, 256);
+		PQfreeCancel(cancel);
+		S->is_running_unbuffered = false;
+	}
+}
+
 static int pgsql_stmt_dtor(pdo_stmt_t *stmt)
 {
 	pdo_pgsql_stmt *S = (pdo_pgsql_stmt*)stmt->driver_data;
@@ -108,6 +119,8 @@ static int pgsql_stmt_dtor(pdo_stmt_t *stmt)
 		S->query = NULL;
 	}
 
+	pgsql_stmt_cancel(S);
+
 	if (S->cursor_name) {
 		if (server_obj_usable) {
 			pdo_pgsql_db_handle *H = S->H;
@@ -137,10 +150,12 @@ static int pgsql_stmt_execute(pdo_stmt_t *stmt)
 	pdo_pgsql_stmt *S = (pdo_pgsql_stmt*)stmt->driver_data;
 	pdo_pgsql_db_handle *H = S->H;
 	ExecStatusType status;
+	int dispatch_result = 1;
 
 	bool in_trans = stmt->dbh->methods->in_transaction(stmt->dbh);
 
 	/* ensure that we free any previous unfetched results */
+	pgsql_stmt_cancel(S);
 	if(S->result) {
 		PQclear(S->result);
 		S->result = NULL;
@@ -219,6 +234,16 @@ static int pgsql_stmt_execute(pdo_stmt_t *stmt)
 				}
 			}
 		}
+		if (S->is_unbuffered) {
+			dispatch_result = PQsendQueryPrepared(H->server, S->stmt_name,
+					stmt->bound_params ?
+						zend_hash_num_elements(stmt->bound_params) :
+						0,
+					(const char**)S->param_values,
+					S->param_lengths,
+					S->param_formats,
+					0);
+		} else {
 		S->result = PQexecPrepared(H->server, S->stmt_name,
 				stmt->bound_params ?
 					zend_hash_num_elements(stmt->bound_params) :
@@ -227,22 +252,51 @@ static int pgsql_stmt_execute(pdo_stmt_t *stmt)
 				S->param_lengths,
 				S->param_formats,
 				0);
+		}
 	} else if (stmt->supports_placeholders == PDO_PLACEHOLDER_NAMED) {
 		/* execute query with parameters */
+		if (S->is_unbuffered) {
+			dispatch_result = PQsendQueryParams(H->server, ZSTR_VAL(S->query),
+					stmt->bound_params ? zend_hash_num_elements(stmt->bound_params) : 0,
+					S->param_types,
+					(const char**)S->param_values,
+					S->param_lengths,
+					S->param_formats,
+					0);
+		} else {
 		S->result = PQexecParams(H->server, ZSTR_VAL(S->query),
 				stmt->bound_params ? zend_hash_num_elements(stmt->bound_params) : 0,
 				S->param_types,
 				(const char**)S->param_values,
 				S->param_lengths,
 				S->param_formats,
 				0);
+		}
 	} else {
 		/* execute plain query (with embedded parameters) */
+		if (S->is_unbuffered) {
+			dispatch_result = PQsendQuery(H->server, ZSTR_VAL(stmt->active_query_string));
+		} else {
 		S->result = PQexec(H->server, ZSTR_VAL(stmt->active_query_string));
+		}
 	}
+
+	if (S->is_unbuffered) {
+		if (!dispatch_result) {
+			pdo_pgsql_error_stmt(stmt, 0, NULL);
+			return 0;
+		}
+		S->is_running_unbuffered = true;
+		PQsetSingleRowMode(H->server);
+		/* no matter if it returns 0: PQ then transparently fallbacks to full result fetching */
+
+		/* try a first fetch to at least have column names and so on */
+		S->result = PQgetResult(S->H->server);
+	}
+
 	status = PQresultStatus(S->result);
 
-	if (status != PGRES_COMMAND_OK && status != PGRES_TUPLES_OK) {
+	if (status != PGRES_COMMAND_OK && status != PGRES_TUPLES_OK && status != PGRES_SINGLE_TUPLE) {
 		pdo_pgsql_error_stmt(stmt, status, pdo_pgsql_sqlstate(S->result));
 		return 0;
 	}
@@ -464,6 +518,35 @@ static int pgsql_stmt_fetch(pdo_stmt_t *stmt,
 			return 0;
 		}
 	} else {
+		if (S->is_running_unbuffered && S->current_row >= stmt->row_count) {
+			ExecStatusType status;
+
+			/* @todo in unbuffered mode, PQ allows multiple queries to be passed:
+			 *       column_count should be recomputed on each iteration */
+
+			if(S->result) {
+				PQclear(S->result);
+				S->result = NULL;
+			}
+
+			S->result = PQgetResult(S->H->server);
+			status = PQresultStatus(S->result);
+
+			if (status != PGRES_COMMAND_OK && status != PGRES_TUPLES_OK && status != PGRES_SINGLE_TUPLE) {
+				pdo_pgsql_error_stmt(stmt, status, pdo_pgsql_sqlstate(S->result));
+				return 0;
+			}
+
+			stmt->row_count = (zend_long)PQntuples(S->result);
+			S->current_row = 0;
+
+			if (!stmt->row_count) {
+				/* libpq requires looping until getResult returns null */
+				PQgetResult(S->H->server);
+				/* @todo receiving a result here is unexpected and should throw an error */
+				S->is_running_unbuffered = false;
+			}
+		}
 		if (S->current_row < stmt->row_count) {
 			S->current_row++;
 			return 1;

ext/pdo_pgsql/php_pdo_pgsql_int.h

@@ -66,6 +66,8 @@ typedef struct {
 	Oid *param_types;
 	int                     current_row;
 	bool is_prepared;
+	bool is_unbuffered;
+	bool is_running_unbuffered;
 } pdo_pgsql_stmt;
 
 typedef struct {