[misc] compression correction for multi-packet

Author: diego Dupin

src/main/java/org/mariadb/jdbc/client/socket/impl/CompressOutputStream.java

@@ -15,6 +15,7 @@ public class CompressOutputStream extends OutputStream {
  private final OutputStream out;
  private final MutableInt sequence;
  private final byte[] header = new byte[7];
+ private byte[] longPacketBuffer = null;
 
  public CompressOutputStream(OutputStream out, MutableInt compressionSequence) {
  this.out = out;
@@ -46,11 +47,26 @@ public CompressOutputStream(OutputStream out, MutableInt compressionSequence) {
  */
  @Override
  public void write(byte[] b, int off, int len) throws IOException {
- if (len < MIN_COMPRESSION_SIZE) {
+ if (len + ((longPacketBuffer != null) ? longPacketBuffer.length : 0) < MIN_COMPRESSION_SIZE) {
  // *******************************************************************************
  // small packet, no compression
  // *******************************************************************************
 
+ if (longPacketBuffer != null) {
+ header[0] = (byte) (len + longPacketBuffer.length);
+ header[1] = (byte) ((len + longPacketBuffer.length) >>> 8);
+ header[2] = 0;
+ header[3] = sequence.incrementAndGet();
+ header[4] = 0;
+ header[5] = 0;
+ header[6] = 0;
+ out.write(header, 0, 7);
+ out.write(longPacketBuffer, 0, longPacketBuffer.length);
+ out.write(b, off, len);
+ longPacketBuffer = null;
+ return;
+ }
+
  header[0] = (byte) len;
  header[1] = (byte) (len >>> 8);
  header[2] = 0;
@@ -62,31 +78,51 @@ public void write(byte[] b, int off, int len) throws IOException {
  out.write(b, off, len);
 
  } else {
+ // *******************************************************************************
+ // compressing packet
+ // *******************************************************************************
+ int sent = 0;
+ try (ByteArrayOutputStream baos = new ByteArrayOutputStream()) {
+ try (DeflaterOutputStream deflater = new DeflaterOutputStream(baos)) {
 
- // *******************************************************************************
- // compressing packet
- // *******************************************************************************
- try (ByteArrayOutputStream baos = new ByteArrayOutputStream()) {
- try (DeflaterOutputStream deflater = new DeflaterOutputStream(baos)) {
- deflater.write(b, off, len);
- deflater.finish();
- }
+ /**
+ * For multi packet, len will be 0x00ffffff + 4 bytes for header.
+ * but compression can only compress up to 0x00ffffff bytes (header initial length size cannot be > 3 bytes)
+ * so, for this specific case, a buffer will be
+ */
 
- byte[] compressedBytes = baos.toByteArray();
+ if (longPacketBuffer != null) {
+ deflater.write(longPacketBuffer, 0, longPacketBuffer.length);
+ sent = longPacketBuffer.length;
+ longPacketBuffer = null;
+ }
+ if ( len + sent > 0x00ffffff) {
+ int remaining = len + sent - 0x00ffffff;
+ longPacketBuffer = new byte[remaining];
+ System.arraycopy(b, off + 0x00ffffff - sent, longPacketBuffer, 0, remaining);
+ }
 
- int compressLen = compressedBytes.length;
+ int bufLenSent = Math.min(0x00ffffff - sent, len);
+ deflater.write(b, off, bufLenSent);
+ sent += bufLenSent;
+ deflater.finish();
+ }
 
- header[0] = (byte) compressLen;
- header[1] = (byte) (compressLen >>> 8);
- header[2] = (byte) (compressLen >>> 16);
- header[3] = sequence.incrementAndGet();
- header[4] = (byte) len;
- header[5] = (byte) (len >>> 8);
- header[6] = (byte) (len >>> 16);
+ byte[] compressedBytes = baos.toByteArray();
 
- out.write(header, 0, 7);
- out.write(compressedBytes, 0, compressLen);
- }
+ int compressLen = compressedBytes.length;
+
+ header[0] = (byte) compressLen;
+ header[1] = (byte) (compressLen >>> 8);
+ header[2] = (byte) (compressLen >>> 16);
+ header[3] = sequence.incrementAndGet();
+ header[4] = (byte) sent;
+ header[5] = (byte) (sent >>> 8);
+ header[6] = (byte) (sent >>> 16);
+
+ out.write(header, 0, 7);
+ out.write(compressedBytes, 0, compressLen);
+ }
  }
  }
 
@@ -107,6 +143,11 @@ public void write(byte[] b, int off, int len) throws IOException {
  */
  @Override
  public void flush() throws IOException {
+ if (longPacketBuffer != null) {
+ byte[] b = longPacketBuffer;
+ longPacketBuffer = null;
+ write(b, 0, b.length);
+ }
  out.flush();
  sequence.set((byte) -1);
  }

src/test/java/org/mariadb/jdbc/integration/CompressTest.java

@@ -10,6 +10,7 @@
 import org.junit.jupiter.api.*;
 import org.mariadb.jdbc.Connection;
 import org.mariadb.jdbc.Statement;
+import org.mariadb.jdbc.util.constants.Capabilities;
 
 public class CompressTest extends Common {
  private static Connection shareCompressCon;
@@ -42,7 +43,7 @@ public void bigSend() throws SQLException {
  }
 
  public void bigSend(Connection con, int maxLen) throws SQLException {
- char[] arr2 = new char[Math.min(maxLen, Math.min(16 * 1024 * 1024, getMaxAllowedPacket() / 2))];
+ char[] arr2 = new char[Math.min(maxLen, Math.min(16 * 1024 * 1024, (getMaxAllowedPacket() / 2) - 1000))];
  for (int pos = 0; pos < arr2.length; pos++) {
  arr2[pos] = (char) ('A' + (pos % 60));
  }

src/test/java/org/mariadb/jdbc/integration/PreparedStatementParametersTest.java

@@ -462,16 +462,21 @@ public void bigSend(Connection con, String st) throws SQLException {
  @Test
  public void bigSendError() throws SQLException {
  int maxAllowedPacket = getMaxAllowedPacket();
- Assumptions.assumeTrue(maxAllowedPacket < 10 * 1024 * 1024);
- char[] arr = new char[10 * 1024 * 1024];
+ Assumptions.assumeTrue(maxAllowedPacket < 32 * 1024 * 1024);
+ char[] arr = new char[maxAllowedPacket];
  for (int pos = 0; pos < arr.length; pos++) {
  arr[pos] = (char) ('A' + (pos % 60));
  }
+ boolean expectClosed = maxAllowedPacket > 16 * 1024 * 1024;
  String st = new String(arr);
- bigSendError(sharedConn, st, true);
- bigSendError(sharedConnBinary, st, true);
+ try (Connection con = createCon()) {
+ bigSendError(con, st, expectClosed);
+ }
+ try (Connection con = createCon("useServerPrepStmts=true")) {
+ bigSendError(con, st, expectClosed);
+ }
  try (Connection con = createCon("transactionReplay")) {
- bigSendError(con, st, false);
+ bigSendError(con, st, expectClosed);
  }
  }