diff --git a/CHANGELOG.md b/CHANGELOG.md index 8c930475c..fbf22c5dd 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/convex-db/src/main/java/convex/db/psql/PgProtocolHandler.java b/convex-db/src/main/java/convex/db/psql/PgProtocolHandler.java index 51ce9a97b..03727ee5b 100644 --- a/convex-db/src/main/java/convex/db/psql/PgProtocolHandler.java +++ b/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. + * + *

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. diff --git a/convex-db/src/test/java/convex/db/psql/PgBinaryParameterTest.java b/convex-db/src/test/java/convex/db/psql/PgBinaryParameterTest.java new file mode 100644 index 000000000..1b09cb994 --- /dev/null +++ b/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 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 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 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; + }); + } +} diff --git a/convex-db/src/test/java/convex/db/psql/PgServerTest.java b/convex-db/src/test/java/convex/db/psql/PgServerTest.java index a86bf6d7c..6b305786b 100644 --- a/convex-db/src/test/java/convex/db/psql/PgServerTest.java +++ b/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 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 {