Pdo\Pgsql: Fix getColumnMeta() for GH-15287 Pdo\Pgsql::setAttribute(PDO::ATTR_PREFETCH, 0)

As stated in the UPGRADING, using the passthrough ("single-row") mode of libpq (introduced in #15287)
forbids passing a new query while the current one's results have not been entirely consumed.
… But I didn't notice that ext/pdo_pgsql internally used new queries to fetch metadata (example use case:
a call to getColumnMeta() while fetch()ing row by row will interleave the getColumnMeta()-triggered
internal query to the database, with the results fetching for the user-called query).

This PR makes those internal calls return NULL for non-essential metadata, instead of letting libpq abort the user-called query.

It moreover includes a small tweak to table oid-to-name translation, with a 1-slot cache.
This may by chance allow the internal call to return something instead of NULL,
but it will nonetheless avoid 30 server calls to get the table name of 30 columns of the same table.

optimize calls to foreach(columns of the same table) getColumnMeta()

- each call queried the DB to know the name associated with the table's OID:
  cache the result between two calls
- make pdo_pgsql_translate_oid_to_table higher-level,
  with the last parameter being the handle instead of the raw connection;
  thus the statement is cleaner, letting the handle do all memory handling on the table oid-to-name translation cache
  (which by the way is a driver feature more than a statement one)

close GH-16249

Author: outtersg

NEWS

@@ -7,4 +7,8 @@ PHP                                                                        NEWS
     with a given skeleton, locale, collapse type and identity fallback.
     (BogdanUngureanu)
 
+- PDO_PGSQL:
+  . Fixed Pdo\Pgsql::getColumnMeta() when Pdo\Pgsql::setAttribute(PDO::ATTR_PREFETCH, 0).
+    (outtersg)
+
 <<< NOTE: Insert NEWS from last stable release here prior to actual release! >>>

ext/pdo_pgsql/pgsql_driver.c

@@ -36,6 +36,7 @@
 #include "pgsql_driver_arginfo.h"
 
 static bool pgsql_handle_in_transaction(pdo_dbh_t *dbh);
+void pgsql_stmt_finish(pdo_pgsql_stmt *S, int fin_mode);
 
 static char * _pdo_pgsql_trim_message(const char *message, int persistent)
 {
@@ -109,6 +110,37 @@ int _pdo_pgsql_error(pdo_dbh_t *dbh, pdo_stmt_t *stmt, int errcode, const char *
 }
 /* }}} */
 
+static zend_always_inline void pgsql_finish_running_stmt(pdo_pgsql_db_handle *H)
+{
+	if (H->running_stmt) {
+		pgsql_stmt_finish(H->running_stmt, 0);
+	}
+}
+
+static zend_always_inline void pgsql_discard_running_stmt(pdo_pgsql_db_handle *H)
+{
+	if (H->running_stmt) {
+		pgsql_stmt_finish(H->running_stmt, FIN_DISCARD);
+	}
+
+	PGresult *pgsql_result;
+	bool first = true;
+	while ((pgsql_result = PQgetResult(H->server))) {
+		/* We should not arrive here, where libpq has a result to deliver without us
+		 * having registered a running statement:
+		 * every result discarding should go through the unified pgsql_stmt_finish,
+		 * but maybe there still is an internal query that we omitted to adapt.
+		 * So instead of asserting let's just emit an informational notice,
+		 * and consume anyway (results consumption is handle-wise, so we have no formal
+		 * need for the statement). */
+		if (first) {
+			php_error_docref(NULL, E_NOTICE, "Internal error: unable to link a libpq result to consume, to its origin statement");
+			first = false;
+		}
+		PQclear(pgsql_result);
+	}
+}
+
 static void _pdo_pgsql_notice(void *context, const char *message) /* {{{ */
 {
 	pdo_dbh_t * dbh = (pdo_dbh_t *)context;
@@ -258,6 +290,10 @@ static void pgsql_handle_closer(pdo_dbh_t *dbh) /* {{{ */
 			PQfinish(H->server);
 			H->server = NULL;
 		}
+		if (H->cached_table_name) {
+			efree(H->cached_table_name);
+			H->cached_table_name = NULL;
+		}
 		if (H->einfo.errmsg) {
 			pefree(H->einfo.errmsg, dbh->is_persistent);
 			H->einfo.errmsg = NULL;
@@ -351,6 +387,7 @@ static zend_long pgsql_handle_doer(pdo_dbh_t *dbh, const zend_string *sql)
 
 	bool in_trans = pgsql_handle_in_transaction(dbh);
 
+	pgsql_finish_running_stmt(H);
 	if (!(res = PQexec(H->server, ZSTR_VAL(sql)))) {
 		/* fatal error */
 		pdo_pgsql_error(dbh, PGRES_FATAL_ERROR, NULL);
@@ -426,6 +463,7 @@ static zend_string *pdo_pgsql_last_insert_id(pdo_dbh_t *dbh, const zend_string *
 	PGresult *res;
 	ExecStatusType status;
 
+	pgsql_finish_running_stmt(H);
 	if (name == NULL) {
 		res = PQexec(H->server, "SELECT LASTVAL()");
 	} else {
@@ -589,6 +627,7 @@ static bool pdo_pgsql_transaction_cmd(const char *cmd, pdo_dbh_t *dbh)
 	PGresult *res;
 	bool ret = true;
 
+	pgsql_finish_running_stmt(H);
 	res = PQexec(H->server, cmd);
 
 	if (PQresultStatus(res) != PGRES_COMMAND_OK) {
@@ -696,9 +735,8 @@ void pgsqlCopyFromArray_internal(INTERNAL_FUNCTION_PARAMETERS)
 	/* Obtain db Handle */
 	H = (pdo_pgsql_db_handle *)dbh->driver_data;
 
-	while ((pgsql_result = PQgetResult(H->server))) {
-		PQclear(pgsql_result);
-	}
+	pgsql_discard_running_stmt(H);
+
 	pgsql_result = PQexec(H->server, query);
 
 	efree(query);
@@ -820,9 +858,8 @@ void pgsqlCopyFromFile_internal(INTERNAL_FUNCTION_PARAMETERS)
 
 	H = (pdo_pgsql_db_handle *)dbh->driver_data;
 
-	while ((pgsql_result = PQgetResult(H->server))) {
-		PQclear(pgsql_result);
-	}
+	pgsql_discard_running_stmt(H);
+
 	pgsql_result = PQexec(H->server, query);
 
 	efree(query);
@@ -916,9 +953,7 @@ void pgsqlCopyToFile_internal(INTERNAL_FUNCTION_PARAMETERS)
 		RETURN_FALSE;
 	}
 
-	while ((pgsql_result = PQgetResult(H->server))) {
-		PQclear(pgsql_result);
-	}
+	pgsql_discard_running_stmt(H);
 
 	/* using pre-9.0 syntax as PDO_pgsql is 7.4+ compatible */
 	if (pg_fields) {
@@ -1007,9 +1042,7 @@ void pgsqlCopyToArray_internal(INTERNAL_FUNCTION_PARAMETERS)
 
 	H = (pdo_pgsql_db_handle *)dbh->driver_data;
 
-	while ((pgsql_result = PQgetResult(H->server))) {
-		PQclear(pgsql_result);
-	}
+	pgsql_discard_running_stmt(H);
 
 	/* using pre-9.0 syntax as PDO_pgsql is 7.4+ compatible */
 	if (pg_fields) {
@@ -1461,6 +1494,7 @@ static int pdo_pgsql_handle_factory(pdo_dbh_t *dbh, zval *driver_options) /* {{{
 
 	H->attached = 1;
 	H->pgoid = -1;
+	H->cached_table_oid = InvalidOid;
 
 	dbh->methods = &pgsql_methods;
 	dbh->alloc_own_columns = 1;

ext/pdo_pgsql/pgsql_statement.c

@@ -56,14 +56,14 @@
 #define FLOAT8LABEL "float8"
 #define FLOAT8OID 701
 
-#define FIN_DISCARD 0x1
-#define FIN_CLOSE   0x2
-#define FIN_ABORT   0x4
 
 
-
-static void pgsql_stmt_finish(pdo_pgsql_stmt *S, int fin_mode)
+void pgsql_stmt_finish(pdo_pgsql_stmt *S, int fin_mode)
 {
+	if (!S) {
+		return;
+	}
+
 	pdo_pgsql_db_handle *H = S->H;
 
 	if (S->is_running_unbuffered && S->result && (fin_mode & FIN_ABORT)) {
@@ -113,9 +113,10 @@ static void pgsql_stmt_finish(pdo_pgsql_stmt *S, int fin_mode)
 		}
 
 		S->is_prepared = false;
-		if (H->running_stmt == S) {
-			H->running_stmt = NULL;
-		}
+	}
+
+	if (H->running_stmt == S && (fin_mode & (FIN_CLOSE|FIN_ABORT))) {
+		H->running_stmt = NULL;
 	}
 }
 
@@ -188,9 +189,8 @@ static int pgsql_stmt_execute(pdo_stmt_t *stmt)
 	 * and returns a PGRES_FATAL_ERROR when PQgetResult gets called for stmt 2 if DEALLOCATE
 	 * was called for stmt 1 inbetween
 	 * (maybe it will change with pipeline mode in libpq 14?) */
-	if (S->is_unbuffered && H->running_stmt) {
+	if (H->running_stmt && H->running_stmt->is_unbuffered) {
 		pgsql_stmt_finish(H->running_stmt, FIN_CLOSE);
-		H->running_stmt = NULL;
 	}
 	/* ensure that we free any previous unfetched results */
 	pgsql_stmt_finish(S, 0);
@@ -702,12 +702,29 @@ static int pgsql_stmt_get_col(pdo_stmt_t *stmt, int colno, zval *result, enum pd
 	return 1;
 }
 
-static zend_always_inline char * pdo_pgsql_translate_oid_to_table(Oid oid, PGconn *conn)
+static zend_always_inline char * pdo_pgsql_translate_oid_to_table(Oid oid, pdo_pgsql_db_handle *H)
 {
+	PGconn *conn = H->server;
 	char *table_name = NULL;
 	PGresult *tmp_res;
 	char *querystr = NULL;
 
+	if (oid == H->cached_table_oid) {
+		return H->cached_table_name;
+	}
+
+	if (H->running_stmt && H->running_stmt->is_unbuffered) {
+		/* in single-row mode, libpq forbids passing a new query
+		 * while we're still flushing the current one's result */
+		return NULL;
+	}
+
+	if (H->cached_table_name) {
+		efree(H->cached_table_name);
+		H->cached_table_name = NULL;
+		H->cached_table_oid = InvalidOid;
+	}
+
 	spprintf(&querystr, 0, "SELECT RELNAME FROM PG_CLASS WHERE OID=%d", oid);
 
 	if ((tmp_res = PQexec(conn, querystr)) == NULL || PQresultStatus(tmp_res) != PGRES_TUPLES_OK) {
@@ -724,6 +741,8 @@ static zend_always_inline char * pdo_pgsql_translate_oid_to_table(Oid oid, PGcon
 		return 0;
 	}
 
+	H->cached_table_oid = oid;
+	H->cached_table_name = estrdup(table_name);
 	table_name = estrdup(table_name);
 
 	PQclear(tmp_res);
@@ -752,10 +771,9 @@ static int pgsql_stmt_get_column_meta(pdo_stmt_t *stmt, zend_long colno, zval *r
 
 	table_oid = PQftable(S->result, colno);
 	add_assoc_long(return_value, "pgsql:table_oid", table_oid);
-	table_name = pdo_pgsql_translate_oid_to_table(table_oid, S->H->server);
+	table_name = pdo_pgsql_translate_oid_to_table(table_oid, S->H);
 	if (table_name) {
-		add_assoc_string(return_value, "table", table_name);
-		efree(table_name);
+		add_assoc_string(return_value, "table", S->H->cached_table_name);
 	}
 
 	switch (S->cols[colno].pgsql_type) {
@@ -794,6 +812,10 @@ static int pgsql_stmt_get_column_meta(pdo_stmt_t *stmt, zend_long colno, zval *r
 			break;
 		default:
 			/* Fetch metadata from Postgres system catalogue */
+			if (S->H->running_stmt && S->H->running_stmt->is_unbuffered) {
+				/* libpq forbids calling a query while we're still reading the preceding one's */
+				break;
+			}
 			spprintf(&q, 0, "SELECT TYPNAME FROM PG_TYPE WHERE OID=%u", S->cols[colno].pgsql_type);
 			res = PQexec(S->H->server, q);
 			efree(q);

ext/pdo_pgsql/php_pdo_pgsql_int.h

@@ -43,6 +43,8 @@ typedef struct {
 	unsigned 	_reserved:31;
 	pdo_pgsql_error_info	einfo;
 	Oid 		pgoid;
+	Oid cached_table_oid;
+	char *cached_table_name;
 	unsigned int	stmt_counter;
 	bool		emulate_prepares;
 	bool		disable_prepares;
@@ -90,6 +92,12 @@ extern int _pdo_pgsql_error(pdo_dbh_t *dbh, pdo_stmt_t *stmt, int errcode, const
 
 extern const struct pdo_stmt_methods pgsql_stmt_methods;
 
+#define FIN_DISCARD 0x1
+#define FIN_CLOSE   0x2
+#define FIN_ABORT   0x4
+
+extern void pgsql_stmt_finish(pdo_pgsql_stmt *S, int fin_mode);
+
 #define pdo_pgsql_sqlstate(r) PQresultErrorField(r, PG_DIAG_SQLSTATE)
 
 enum {

ext/pdo_pgsql/tests/gh15287.phpt

@@ -132,6 +132,17 @@ $res = []; while (($re = $stmt->fetch())) $res[] = $re; display($res);
 $stmt->execute([ 0 ]);
 $res = []; for ($i = -1; ++$i < 2;) $res[] = $stmt->fetch(); display($res);
 display($pdo->query("select * from t2")->fetchAll());
+
+// Metadata calls the server for some operations (notably table oid-to-name conversion).
+// This will break libpq (that forbids a second PQexec before we consumed the first one).
+// Instead of either letting libpq return an error, or blindly forbid this call, we expect
+// being transparently provided at least attributes which do not require a server roundtrip.
+// And good news: column name is one of those "local" attributes.
+echo "=== meta ===\n";
+$stmt = $pdo->query("select * from t limit 2");
+echo "Starting with column " . $stmt->getColumnMeta(0)['name'] . ":\n";
+display($stmt->fetchAll());
+
 ?>
 --EXPECTF--
 === non regression ===
@@ -181,3 +192,7 @@ multiple calls to the same prepared statement, some interrupted before having re
 0
 1
 678	ok
+=== meta ===
+Starting with column n:
+0	original
+1	non original