Fix passing variable types to postgres

drift/lib/src/runtime/query_builder/expressions/variables.dart

@@ -76,7 +76,7 @@ class Variable<T extends Object> extends Expression<T> {
     var mark = '?';
     if (context.dialect == SqlDialect.postgres) {
       explicitStart = 1;
-      mark = '@';
+      mark = r'$';
     }
 
     if (explicitStart != null) {

extras/drift_postgres/lib/src/pg_database.dart

@@ -60,21 +60,28 @@ class _PgDelegate extends DatabaseDelegate {
   @override
   Future<void> runBatched(BatchedStatements statements) async {
     final session = _openedSession!;
-    final prepared = <PgStatement>[];
+    final prepared =
+        List<PgStatement?>.filled(statements.statements.length, null);
 
     try {
-      for (final statement in statements.statements) {
-        prepared.add(await session.prepare(PgSql(statement)));
-      }
+      for (final instantation in statements.arguments) {
+        final pgArgs = _BoundArguments.ofDartArgs(instantation.arguments);
 
-      for (final row in statements.arguments) {
-        final stmt = prepared[row.statementIndex];
+        // Lazily prepare statements when we run into them. The reason is that
+        // we need to know the types for variables.
+        final stmtIndex = instantation.statementIndex;
+        var stmt = prepared[stmtIndex];
+        if (stmt == null) {
+          final sql = statements.statements[stmtIndex];
+          stmt = prepared[stmtIndex] =
+              await session.prepare(PgSql(sql, types: pgArgs.types));
+        }
 
-        await stmt.run(row.arguments);
+        await stmt.run(pgArgs.parameters);
       }
     } finally {
       for (final stmt in prepared) {
-        await stmt.dispose();
+        await stmt?.dispose();
       }
     }
   }
@@ -82,7 +89,11 @@ class _PgDelegate extends DatabaseDelegate {
   Future<int> _runWithArgs(String statement, List<Object?> args) async {
     final session = _openedSession!;
 
-    final result = await session.execute(PgSql(statement), parameters: args);
+    final pgArgs = _BoundArguments.ofDartArgs(args);
+    final result = await session.execute(
+      PgSql(statement, types: pgArgs.types),
+      parameters: pgArgs.parameters,
+    );
     return result.affectedRows;
   }
 
@@ -94,7 +105,9 @@ class _PgDelegate extends DatabaseDelegate {
   @override
   Future<int> runInsert(String statement, List<Object?> args) async {
     final session = _openedSession!;
-    final result = await session.execute(PgSql(statement), parameters: args);
+    final pgArgs = _BoundArguments.ofDartArgs(args);
+    final result = await session.execute(PgSql(statement, types: pgArgs.types),
+        parameters: pgArgs.parameters);
     return result.firstOrNull?[0] as int? ?? 0;
   }
 
@@ -106,7 +119,9 @@ class _PgDelegate extends DatabaseDelegate {
   @override
   Future<QueryResult> runSelect(String statement, List<Object?> args) async {
     final session = _openedSession!;
-    final result = await session.execute(PgSql(statement), parameters: args);
+    final pgArgs = _BoundArguments.ofDartArgs(args);
+    final result = await session.execute(PgSql(statement, types: pgArgs.types),
+        parameters: pgArgs.parameters);
 
     return QueryResult([
       for (final pgColumn in result.schema.columns) pgColumn.columnName ?? '',
@@ -119,12 +134,46 @@ class _PgDelegate extends DatabaseDelegate {
       await _openedSession?.close();
     }
   }
+}
 
-  Object? _convertValue(Object? value) {
-    if (value is BigInt) {
-      return value.toInt();
+class _BoundArguments {
+  final List<PgDataType> types;
+  final List<PgTypedParameter> parameters;
+
+  _BoundArguments(this.types, this.parameters);
+
+  factory _BoundArguments.ofDartArgs(List<Object?> args) {
+    final types = <PgDataType>[];
+    final parameters = <PgTypedParameter>[];
+
+    void add(PgTypedParameter param) {
+      types.add(param.type);
+      parameters.add(param);
     }
-    return value;
+
+    for (final value in args) {
+      if (value == null) {
+        add(PgTypedParameter(PgDataType.text, null));
+      } else if (value is int) {
+        add(PgTypedParameter(PgDataType.bigInteger, value));
+      } else if (value is BigInt) {
+        // Drift only uses BigInts to represent 64-bit values on the web, so we
+        // can use toInt() here.
+        add(PgTypedParameter(PgDataType.bigInteger, value));
+      } else if (value is bool) {
+        add(PgTypedParameter(PgDataType.boolean, value));
+      } else if (value is double) {
+        add(PgTypedParameter(PgDataType.double, value));
+      } else if (value is String) {
+        add(PgTypedParameter(PgDataType.text, value));
+      } else if (value is List<int>) {
+        add(PgTypedParameter(PgDataType.byteArray, value));
+      } else {
+        throw ArgumentError.value(value, 'value', 'Unsupported type');
+      }
+    }
+
+    return _BoundArguments(types, parameters);
   }
 }