Decode binary-format parameters by their declared type in PgProtocolHandler

CHANGELOG.md

@@ -24,6 +24,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- PostgreSQL prepared statements decode binary numeric parameters by their declared type, including the full unsigned OID range, and reject invalid fixed-width parameter lengths.
 - Scheduled execution respects the 100-operation limit across timestamp buckets and retains unprocessed entries, preventing oversized buckets from exhausting peer memory.
 - Variadic subtraction reports a numeric error when an intermediate result exceeds the integer limit, instead of throwing a Java exception.
 - HTTP message requests use independent correlation IDs on the shared peer connection, and rejecting a duplicate client request ID no longer discards the original pending response.

convex-db/src/main/java/convex/db/psql/PgProtocolHandler.java

@@ -540,7 +540,7 @@ private void handleExecute(ChannelHandlerContext ctx, PgMessageDecoder.Execute e
 				return;
 			}
 
-			executeWithParameters(ctx, query, portal.paramValues(), portal.paramFormats());
+			executeWithParameters(ctx, query, portal.paramValues(), portal.paramFormats(), portal.stmt().paramTypes());
 		} catch (SQLException e) {
 			log.warn("Execute error: {}", e.getMessage(), e);
 			write(ctx, ErrorResponse.fromException(e));
@@ -557,7 +557,7 @@ private void handleExecute(ChannelHandlerContext ctx, PgMessageDecoder.Execute e
 	/**
 	 * Execute a query with bound parameters.
 	 */
-	private void executeWithParameters(ChannelHandlerContext ctx, String sql, byte[][] paramValues, short[] paramFormats) throws SQLException {
+	private void executeWithParameters(ChannelHandlerContext ctx, String sql, byte[][] paramValues, short[] paramFormats, int[] paramTypes) throws SQLException {
 		sql = rewriteQuery(sql);
 
 		if (sql == null) {
@@ -603,8 +603,9 @@ private void executeWithParameters(ChannelHandlerContext ctx, String sql, byte[]
 						String strValue = new String(value, java.nio.charset.StandardCharsets.UTF_8);
 						pstmt.setString(i + 1, strValue);
 					} else {
-						// Binary format - set as bytes
-						pstmt.setBytes(i + 1, value);
+						// Binary format - decode by the type the client declared in Parse
+						int oid = (paramTypes != null && i < paramTypes.length) ? paramTypes[i] : 0;
+						bindBinaryParameter(pstmt, i + 1, value, oid);
 					}
 				}
 			}
@@ -630,6 +631,38 @@ private void executeWithParameters(ChannelHandlerContext ctx, String sql, byte[]
 		}
 	}
 
+	/**
+	 * Binds a parameter that arrived in binary format, decoding it by the type OID the
+	 * client declared for it in the Parse message.
+	 *
+	 * <p>The PostgreSQL JDBC driver sends numbers in binary format: setLong arrives as
+	 * the 8 bytes of an int8, not as the digits of the number. Binding those bytes as
+	 * they are hands Calcite a byte string where it expects a number, and the statement
+	 * fails. Types without a decoding here, and parameters the client declared no type
+	 * for, are bound as bytes as before.
+	 */
+	static void bindBinaryParameter(PreparedStatement pstmt, int index, byte[] value, int oid) throws SQLException {
+		java.nio.ByteBuffer buf = java.nio.ByteBuffer.wrap(value);
+		switch (oid) {
+			case PgType.INT2 -> { requireLength(value, 2, "int2"); pstmt.setLong(index, buf.getShort()); }
+			case PgType.INT4 -> { requireLength(value, 4, "int4"); pstmt.setLong(index, buf.getInt()); }
+			case PgType.OID -> { requireLength(value, 4, "oid"); pstmt.setLong(index, Integer.toUnsignedLong(buf.getInt())); }
+			case PgType.INT8 -> { requireLength(value, 8, "int8"); pstmt.setLong(index, buf.getLong()); }
+			case PgType.FLOAT4 -> { requireLength(value, 4, "float4"); pstmt.setDouble(index, buf.getFloat()); }
+			case PgType.FLOAT8 -> { requireLength(value, 8, "float8"); pstmt.setDouble(index, buf.getDouble()); }
+			case PgType.BOOL -> { requireLength(value, 1, "bool"); pstmt.setBoolean(index, value[0] != 0); }
+			case PgType.TEXT, PgType.VARCHAR, PgType.CHAR, PgType.NAME ->
+				pstmt.setString(index, new String(value, java.nio.charset.StandardCharsets.UTF_8));
+			default -> pstmt.setBytes(index, value);
+		}
+	}
+
+	private static void requireLength(byte[] value, int expected, String type) throws SQLException {
+		if (value.length != expected) {
+			throw new SQLException("Binary " + type + " parameter has " + value.length + " bytes, expected " + expected, "08P01");
+		}
+	}
+
 	/**
 	 * Adds pg_catalog. prefix to a table name if not already qualified.
 	 * PostgreSQL's search path includes pg_catalog, but Calcite requires explicit schema.

convex-db/src/test/java/convex/db/psql/PgBinaryParameterTest.java

@@ -0,0 +1,57 @@
+package convex.db.psql;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.lang.reflect.Proxy;
+import java.nio.ByteBuffer;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.jupiter.api.Test;
+
+/** Tests the values handed to JDBC after PostgreSQL binary decoding. */
+public class PgBinaryParameterTest {
+	@Test
+	public void testUnsignedOid() throws SQLException {
+		for (long value : new long[] {0, 1, 0x7fffffffL, 0x80000000L, 0xffffffffL}) {
+			assertBoundLong(PgType.OID, (int)value, value);
+		}
+	}
+
+	@Test
+	public void testSignedInt4() throws SQLException {
+		for (int value : new int[] {Integer.MIN_VALUE, -1, 0, Integer.MAX_VALUE}) {
+			assertBoundLong(PgType.INT4, value, value);
+		}
+	}
+
+	@Test
+	public void testInvalidOidLengths() {
+		for (int length : new int[] {0, 1, 3, 5, 8}) {
+			AtomicReference<Long> bound=new AtomicReference<>();
+			PreparedStatement statement=captureLong(bound);
+			SQLException error=assertThrows(SQLException.class,
+				() -> PgProtocolHandler.bindBinaryParameter(statement, 7, new byte[length], PgType.OID));
+			assertEquals("08P01",error.getSQLState());
+			assertNull(bound.get(),"Invalid input must not be bound");
+		}
+	}
+
+	private static void assertBoundLong(int oid, int wireValue, long expected) throws SQLException {
+		AtomicReference<Long> bound=new AtomicReference<>();
+		byte[] bytes=ByteBuffer.allocate(4).putInt(wireValue).array();
+		PgProtocolHandler.bindBinaryParameter(captureLong(bound), 7, bytes, oid);
+		assertEquals(Long.valueOf(expected),bound.get());
+	}
+
+	private static PreparedStatement captureLong(AtomicReference<Long> bound) {
+		return (PreparedStatement)Proxy.newProxyInstance(PgBinaryParameterTest.class.getClassLoader(),
+			new Class<?>[] {PreparedStatement.class}, (proxy, method, args) -> {
+				assertEquals("setLong",method.getName());
+				assertEquals(7,args[0]);
+				bound.set((Long)args[1]);
+				return null;
+			});
+	}
+}

convex-db/src/test/java/convex/db/psql/PgServerTest.java

@@ -335,6 +335,93 @@ public void testTerminate() throws IOException {
 	 * Tests connectivity using the PostgreSQL JDBC driver.
 	 * Disabled: Requires extended query protocol support (Parse/Bind/Execute).
 	 */
+	/**
+	 * The PostgreSQL JDBC driver sends numeric parameters in binary format (setLong as
+	 * the 8 bytes of an int8). They must reach the statement as numbers.
+	 */
+	@Test
+	public void testPreparedStatementWithNumericParameters() throws Exception {
+		String url = "jdbc:postgresql://localhost:" + server.getPort() + "/" + dbName + "?user=testuser";
+		try (Connection conn = DriverManager.getConnection(url)) {
+			try (java.sql.PreparedStatement insert = conn.prepareStatement("INSERT INTO users VALUES (?, ?, ?)")) {
+				insert.setLong(1, 3L);
+				insert.setString(2, "Carol");
+				insert.setString(3, "carol@example.com");
+				assertEquals(1, insert.executeUpdate());
+
+				insert.setInt(1, 4);
+				insert.setString(2, "Dave");
+				insert.setString(3, "dave@example.com");
+				assertEquals(1, insert.executeUpdate());
+
+				insert.setShort(1, (short) 5);
+				insert.setString(2, "Erin");
+				insert.setString(3, "erin@example.com");
+				assertEquals(1, insert.executeUpdate());
+
+				// Past the driver's prepare threshold the statement becomes a named
+				// server-side one; the parameters must still arrive as numbers
+				for (int id = 6; id <= 15; id++) {
+					insert.setLong(1, id);
+					insert.setString(2, "User" + id);
+					insert.setString(3, "user" + id + "@example.com");
+					assertEquals(1, insert.executeUpdate());
+				}
+			}
+			try (java.sql.PreparedStatement update = conn.prepareStatement("UPDATE users SET name = ? WHERE id = ?")) {
+				update.setString(1, "Caroline");
+				update.setLong(2, 3L);
+				assertEquals(1, update.executeUpdate());
+			}
+
+		}
+
+		// Read back through the table API: the rows are stored under numeric keys
+		String[] expected = {"Caroline", "Dave", "Erin"};
+		for (int i = 0; i < expected.length; i++) {
+			assertEquals(expected[i], nameOf(3 + i));
+		}
+		for (int id = 6; id <= 15; id++) {
+			assertEquals("User" + id, nameOf(id));
+		}
+		assertEquals(15, db.tables().selectAll("users").count());
+	}
+
+	private String nameOf(long id) {
+		convex.core.data.AVector<convex.core.data.ACell> row = db.tables().selectByKey("users", convex.core.data.prim.CVMLong.create(id));
+		assertNotNull(row, "row " + id);
+		return row.get(1).toString();
+	}
+
+	/**
+	 * Floating point parameters arrive in binary format too (setDouble as a float8,
+	 * setFloat as a float4).
+	 */
+	@Test
+	public void testPreparedStatementWithFloatingPointParameters() throws Exception {
+		ConvexColumnType[] types = {
+			ConvexColumnType.of(ConvexType.INTEGER),
+			ConvexColumnType.of(ConvexType.DOUBLE)
+		};
+		db.tables().createTable("readings", new String[]{"id", "amount"}, types);
+
+		String url = "jdbc:postgresql://localhost:" + server.getPort() + "/" + dbName + "?user=testuser";
+		try (Connection conn = DriverManager.getConnection(url)) {
+			try (java.sql.PreparedStatement insert = conn.prepareStatement("INSERT INTO readings VALUES (?, ?)")) {
+				insert.setLong(1, 1L);
+				insert.setDouble(2, 2.5);
+				assertEquals(1, insert.executeUpdate());
+
+				insert.setLong(1, 2L);
+				insert.setFloat(2, 0.25f);
+				assertEquals(1, insert.executeUpdate());
+			}
+		}
+
+		assertEquals(convex.core.data.prim.CVMDouble.create(2.5), db.tables().selectByKey("readings", convex.core.data.prim.CVMLong.create(1)).get(1));
+		assertEquals(convex.core.data.prim.CVMDouble.create(0.25), db.tables().selectByKey("readings", convex.core.data.prim.CVMLong.create(2)).get(1));
+	}
+
 	@org.junit.jupiter.api.Disabled("Requires extended query protocol")
 	@Test
 	public void testPostgresJDBCDriver() throws Exception {