Let the vertx-pg-client encoder estimate the capacity of buffer it allocates.

Motivation:

The pg encoder allocates Netty buffer without an initial capacity, when pipelining is used this buffer can be reallocated multiple times and in agressive scenario can lead to out of memory errors with the adapative allocator.

Changes:

Cumulate and encode all the outbound messages at once when flush happens instead of cumulating them in the outbound buffer.

Capacity is estimated when a message is enqueued.

At flush time, a single buffer of the estimated capacity is created and the outbound messages are written to this buffer.

This requires to pre-render some messages to estimate their length, e.g. a json object is pre-rendered to a string that can be then estimated. The client has been modified to handle message preparation which takes care of this and lazy duplicate the tuple / list of tuples when this happens.

Author: vietj

vertx-db2-client/src/main/java/io/vertx/db2client/impl/codec/DB2ParamDesc.java

@@ -15,6 +15,7 @@
  */
 package io.vertx.db2client.impl.codec;
 
+import io.vertx.core.VertxException;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.db2client.impl.drda.ClientTypes;
 import io.vertx.db2client.impl.drda.ColumnMetaData;
@@ -35,19 +36,19 @@ ColumnMetaData paramDefinitions() {
     return paramDefinitions;
   }
 
-  public String prepare(TupleInternal values) {
+  public TupleInternal prepare(TupleInternal values) {
     if (values.size() != paramDefinitions.columns_) {
-      return ErrorMessageFactory.buildWhenArgumentsLengthNotMatched(paramDefinitions.columns_, values.size());
+      throw new VertxException(ErrorMessageFactory.buildWhenArgumentsLengthNotMatched(paramDefinitions.columns_, values.size()), true);
     }
     for (int i = 0; i < paramDefinitions.columns_; i++) {
       Object val = values.getValue(i);
       int type = paramDefinitions.types_[i];
       if (!canConvert(val, type)) {
         Class<?> preferredType = ClientTypes.preferredJavaType(type);
-        return ErrorMessageFactory.buildWhenArgumentsTypeNotMatched(preferredType, i, val);
+        throw new VertxException(ErrorMessageFactory.buildWhenArgumentsTypeNotMatched(preferredType, i, val), true);
       }
     }
-    return null;
+    return values;
   }
 
   private static boolean canConvert(Object val, int type) {

vertx-db2-client/src/main/java/io/vertx/db2client/impl/codec/DB2PreparedStatement.java

@@ -74,7 +74,7 @@ public String sql() {
   }
 
   @Override
-  public String prepare(TupleInternal values) {
+  public TupleInternal prepare(TupleInternal values) {
     return paramDesc.prepare(values);
   }
 

vertx-mssql-client/src/main/java/io/vertx/mssqlclient/impl/codec/MSSQLPreparedStatement.java

@@ -40,8 +40,4 @@ public String sql() {
     return sql;
   }
 
-  @Override
-  public String prepare(TupleInternal values) {
-    return null;
-  }
 }

vertx-mysql-client/src/main/java/io/vertx/mysqlclient/impl/codec/MySQLPreparedStatement.java

@@ -17,6 +17,7 @@
 
 package io.vertx.mysqlclient.impl.codec;
 
+import io.vertx.core.VertxException;
 import io.vertx.mysqlclient.impl.MySQLParamDesc;
 import io.vertx.mysqlclient.impl.MySQLRowDesc;
 import io.vertx.mysqlclient.impl.datatype.DataType;
@@ -71,13 +72,13 @@ public String sql() {
   }
 
   @Override
-  public String prepare(TupleInternal values) {
+  public TupleInternal prepare(TupleInternal values) {
     int numberOfParameters = values.size();
     int paramDescLength = paramDesc.paramDefinitions().length;
     if (numberOfParameters != paramDescLength) {
-      return ErrorMessageFactory.buildWhenArgumentsLengthNotMatched(paramDescLength, numberOfParameters);
+      throw new VertxException(ErrorMessageFactory.buildWhenArgumentsLengthNotMatched(paramDescLength, numberOfParameters), true);
     } else {
-      return null;
+      return values;
     }
   }
 

vertx-oracle-client/src/main/java/io/vertx/oracleclient/impl/commands/OraclePreparedStatement.java

@@ -50,10 +50,4 @@ public RowDesc rowDesc() {
   public String sql() {
     return sql;
   }
-
-  @Override
-  public String prepare(TupleInternal values) {
-    return null;
-  }
-
 }

vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/Bind.java

@@ -20,7 +20,7 @@
 /**
  * @author <a href="mailto:[email protected]">Emad Alblueshi</a>
  */
-final class Bind {
+final class Bind extends OutboundMessage {
 
   final byte[] statement;
   final DataType[] paramTypes;

vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ClosePortalMessage.java

@@ -0,0 +1,7 @@
+package io.vertx.pgclient.impl.codec;
+
+class ClosePortalMessage extends OutboundMessage {
+
+  static final ClosePortalMessage INSTANCE = new ClosePortalMessage();
+
+}

vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ClosePreparedStatementMessage.java

@@ -0,0 +1,7 @@
+package io.vertx.pgclient.impl.codec;
+
+class ClosePreparedStatementMessage extends OutboundMessage {
+
+  static final ClosePreparedStatementMessage INSTANCE = new ClosePreparedStatementMessage();
+
+}

vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/DataType.java

@@ -154,14 +154,14 @@ static void encodeText(DataType id, Object value, ByteBuf buff) {
   private static void textEncode(DataType id, Object value, ByteBuf buff) {
     switch (id) {
       case NUMERIC:
-        textEncodeNUMERIC((Number) value, buff);
+        textEncodeNUMERIC((String) value, buff);
         break;
       case NUMERIC_ARRAY:
-        textEncodeNUMERIC_ARRAY((Number[]) value, buff);
+        textEncodeNUMERIC_ARRAY((Object[]) value, buff);
         break;
       case UNKNOWN:
         //default to treating unknown as a string
-        buff.writeCharSequence(String.valueOf(value), StandardCharsets.UTF_8);
+        buff.writeCharSequence((CharSequence) value, StandardCharsets.UTF_8);
         break;
       default:
         logger.debug("Data type " + id + " does not support text encoding");
@@ -281,13 +281,13 @@ public static void encodeBinary(DataType id, Object value, ByteBuf buff) {
         binaryEncodeArray((UUID[]) value, DataType.UUID, buff);
         break;
       case JSON:
-        binaryEncodeJSON((Object) value, buff);
+        binaryEncodeJSON((CharSequence) value, buff);
         break;
       case JSON_ARRAY:
         binaryEncodeArray((Object[]) value, DataType.JSON, buff);
         break;
       case JSONB:
-        binaryEncodeJSONB((Object) value, buff);
+        binaryEncodeJSONB((CharSequence) value, buff);
         break;
       case JSONB_ARRAY:
         binaryEncodeArray((Object[]) value, DataType.JSONB, buff);
@@ -938,12 +938,11 @@ private static Interval textDecodeINTERVAL(int index, int len, ByteBuf buff) {
     return new Interval(years, months, days, hours, minutes, seconds, microseconds);
   }
 
-  private static void textEncodeNUMERIC(Number value, ByteBuf buff) {
-    String s = value.toString();
-    buff.writeCharSequence(s, StandardCharsets.UTF_8);
+  private static void textEncodeNUMERIC(String value, ByteBuf buff) {
+    buff.writeCharSequence(value, StandardCharsets.UTF_8);
   }
 
-  private static void textEncodeNUMERIC_ARRAY(Number[] value, ByteBuf buff) {
+  private static void textEncodeNUMERIC_ARRAY(Object[] value, ByteBuf buff) {
     textEncodeArray(value, DataType.NUMERIC, buff);
   }
 
@@ -1366,14 +1365,8 @@ private static Object binaryDecodeJSON(int index, int len, ByteBuf buff) {
     return textDecodeJSONB(index, len, buff);
   }
 
-  private static void binaryEncodeJSON(Object value, ByteBuf buff) {
-    String s;
-    if (value == Tuple.JSON_NULL) {
-      s = "null";
-    } else {
-      s = Json.encode(value);
-    }
-    buff.writeCharSequence(s, StandardCharsets.UTF_8);
+  private static void binaryEncodeJSON(CharSequence value, ByteBuf buff) {
+    buff.writeCharSequence(value, StandardCharsets.UTF_8);
   }
 
   private static Object textDecodeJSONB(int index, int len, ByteBuf buff) {
@@ -1410,7 +1403,7 @@ private static Object binaryDecodeJSONB(int index, int len, ByteBuf buff) {
     return textDecodeJSONB(index + 1, len - 1, buff);
   }
 
-  private static void binaryEncodeJSONB(Object value, ByteBuf buff) {
+  private static void binaryEncodeJSONB(CharSequence value, ByteBuf buff) {
     buff.writeByte(1); // version
     binaryEncodeJSON(value, buff);
   }