feat: allow explicit use of Simple Query Protocol (v3) (#118)

Author: osaxma

lib/postgres_v3_experimental.dart

@@ -129,11 +129,17 @@ abstract class PgSession {
   /// optimization can be applied also depends on the parameters chosen, so
   /// there is no guarantee that the [PgResult] from a [ignoreRows] excution has
   /// no rows.
+  /// 
+  /// [queryMode] is optional to override the default query execution mode that 
+  /// is defined in [PgSessionSettings]. Unless necessary, always prefer using 
+  /// [QueryMode.extended] which is the default value. For more information,
+  /// see [PgSessionSettings.queryMode]
   Future<PgResult> execute(
     Object /* String | PgSql */ query, {
     Object? /* List<Object?|PgTypedParameter> | Map<String, Object?|PgTypedParameter> */
         parameters,
     bool ignoreRows = false,
+    QueryMode? queryMode,
   });
 
   /// Closes this session, cleaning up resources and forbiding further calls to
@@ -376,13 +382,23 @@ final class PgSessionSettings {
   /// [Streaming Replication Protocol]: https://www.postgresql.org/docs/current/protocol-replication.html
   final ReplicationMode replicationMode;
 
+  /// The Query Execution Mode 
+  /// 
+  /// The default value is [QueryMode.extended] which uses the Extended Query
+  /// Protocol. In certain cases, the Extended protocol cannot be used 
+  /// (e.g. in replication mode or with proxies such as PGBouncer), hence the
+  /// the Simple one would be the only viable option. Unless necessary, always 
+  /// prefer using [QueryMode.extended].
+  final QueryMode queryMode;
+
   PgSessionSettings({
     this.connectTimeout,
     this.timeZone,
     this.encoding,
     this.onBadSslCertificate,
     this.transformer,
     this.replicationMode = ReplicationMode.none,
+    this.queryMode = QueryMode.extended,
   });
 }
 
@@ -393,3 +409,11 @@ final class PgPoolSettings {
     this.maxConnectionCount,
   });
 }
+
+/// Options for the Query Execution Mode
+enum QueryMode {
+  /// Extended Query Protocol
+  extended,
+  /// Simple Query Protocol 
+  simple,
+}

lib/src/text_codec.dart

@@ -242,7 +242,12 @@ class PostgresTextDecoder<T extends Object> {
       case PgDataType.double:
         return num.parse(asText) as T;
       case PgDataType.boolean:
-        return (asText == 'true') as T;
+        // In text data format when using simple query protocol, "true" & "false"
+        // are represented as `t` and `f`,  respectively. 
+        // we will check for both just in case
+        // TODO: should we check for other representations (e.g. `1`, `on`, `y`,
+        // and `yes`)?
+        return (asText == 't' || asText == 'true') as T;
 
       // We could list out all cases, but it's about 20 lines of code.
       // ignore: no_default_cases

lib/src/v3/connection.dart

@@ -42,6 +42,7 @@ class _ResolvedSettings {
   final Encoding encoding;
 
   final ReplicationMode replicationMode;
+  final QueryMode queryMode;
 
   final StreamChannelTransformer<BaseMessage, BaseMessage>? transformer;
 
@@ -56,7 +57,8 @@ class _ResolvedSettings {
         timeZone = settings?.timeZone ?? 'UTC',
         encoding = settings?.encoding ?? utf8,
         transformer = settings?.transformer,
-        replicationMode = settings?.replicationMode ?? ReplicationMode.none;
+        replicationMode = settings?.replicationMode ?? ReplicationMode.none,
+        queryMode = settings?.queryMode ?? QueryMode.extended;
 
   bool onBadSslCertificate(X509Certificate certificate) {
     return settings?.onBadSslCertificate?.call(certificate) ?? false;
@@ -120,36 +122,49 @@ abstract class _PgSessionBase implements PgSession {
     Object query, {
     Object? parameters,
     bool ignoreRows = false,
+    QueryMode? queryMode,
   }) async {
     final description = InternalQueryDescription.wrap(query);
     final variables = description.bindParameters(parameters);
 
-    if (!ignoreRows || variables.isNotEmpty) {
-      // The simple query protocol does not support variables and returns rows
-      // as text. So when we need rows or parameters, we need an explicit prepare.
-      final prepared = await prepare(description);
-      try {
-        return await prepared.run(variables);
-      } finally {
-        await prepared.dispose();
-      }
+    late final bool isSimple;
+    if (queryMode != null) {
+      isSimple = queryMode == QueryMode.simple;
     } else {
+      isSimple = _connection._settings.queryMode == QueryMode.simple;
+    }
+
+    if (isSimple && variables.isNotEmpty) {
+      throw PostgreSQLException('Parameterized queries are not supported when '
+      'using the Simple Query Protocol');
+    }
+
+    if (isSimple || (ignoreRows && variables.isEmpty)) {
       // Great, we can just run a simple query.
       final controller = StreamController<PgResultRow>();
       final items = <PgResultRow>[];
 
-      final querySubscription =
-          _PgResultStreamSubscription.simpleQueryAndIgnoreRows(
+      final querySubscription = _PgResultStreamSubscription.simpleQueryProtocol(
         description.transformedSql,
         this,
         controller,
         controller.stream.listen(items.add),
+        ignoreRows,
       );
       await querySubscription.asFuture();
       await querySubscription.cancel();
 
       return PgResult(items, await querySubscription.affectedRows,
           await querySubscription.schema);
+    } else {
+      // The simple query protocol does not support variables. So when we have
+      // parameters, we need an explicit prepare.
+      final prepared = await prepare(description);
+      try {
+        return await prepared.run(variables);
+      } finally {
+        await prepared.dispose();
+      }
     }
   }
 
@@ -496,9 +511,8 @@ class _PgResultStreamSubscription
     });
   }
 
-  _PgResultStreamSubscription.simpleQueryAndIgnoreRows(
-      String sql, this.session, this._controller, this._source)
-      : ignoreRows = true {
+  _PgResultStreamSubscription.simpleQueryProtocol(String sql, this.session,
+      this._controller, this._source, this.ignoreRows) {
     session._withResource(() async {
       connection._pending = this;
 

lib/src/v3/pool.dart

@@ -42,12 +42,17 @@ class PoolImplementation implements PgPool {
   }
 
   @override
-  Future<PgResult> execute(Object query,
-      {Object? parameters, bool ignoreRows = false}) {
+  Future<PgResult> execute(
+    Object query, {
+    Object? parameters,
+    bool ignoreRows = false,
+    QueryMode? queryMode,
+  }) {
     return withConnection((connection) => connection.execute(
           query,
           parameters: parameters,
           ignoreRows: ignoreRows,
+          queryMode: queryMode,
         ));
   }
 
@@ -161,12 +166,17 @@ class _PoolConnection implements PgConnection {
   }
 
   @override
-  Future<PgResult> execute(Object query,
-      {Object? parameters, bool ignoreRows = false}) {
+  Future<PgResult> execute(
+    Object query, {
+    Object? parameters,
+    bool ignoreRows = false,
+    QueryMode? queryMode,
+  }) {
     return _connection.execute(
       query,
       parameters: parameters,
       ignoreRows: ignoreRows,
+      queryMode: queryMode,
     );
   }
 

test/v3_test.dart

@@ -402,6 +402,29 @@ void main() {
         expect(await connection.execute('SELECT id FROM t'), isEmpty);
       });
     });
+
+    group('Simple Query Protocol', () {
+      test('single simple query', () async {
+        final res = await connection.execute(
+          "SELECT 'dart', 42, true, false, NULL",
+          queryMode: QueryMode.simple,
+        );
+        expect(res, [
+          ['dart', 42, true, false, null]
+        ]);
+      });
+
+      test('parameterized query throws', () async {
+        await expectLater(
+          () => connection.execute(
+            r'SELECT 1',
+            parameters: [PgTypedParameter(PgDataType.integer, 1)],
+            queryMode: QueryMode.simple
+          ),
+          _throwsPostgresException,
+        );
+      });
+    });
   });
 
   test('can inject transformer into connection', () async {