[misc] bulk batching multiple packet correction for parameters bigger than 16M

Author: rusher

.travis/script.sh

@@ -76,10 +76,6 @@ if [ -z "$SKYSQL" ] && [ -z "$SKYSQL_HA" ]; then
  echo "Running tests for JDK version: $TRAVIS_JDK_VERSION"
  mvn clean test $ADDITIONNAL_VARIABLES -DjobId=${TRAVIS_JOB_ID}
 
- if ![ $? -eq 0 ]
- then
- docker-compose -f ${COMPOSE_FILE} exec ${DB} tail -n 1000 /var/log/syslog
- fi
  fi
 else
  if [ -n "$SKYSQL" ]; then

src/main/java/org/mariadb/jdbc/Configuration.java

@@ -241,7 +241,10 @@ private Configuration(
  rewriteBatchedStatements != null ? rewriteBatchedStatements : false;
  this.useCompression = useCompression != null ? useCompression : false;
  this.blankTableNameMeta = blankTableNameMeta != null ? blankTableNameMeta : false;
- this.sslMode = sslMode != null ? SslMode.from(sslMode) : SslMode.DISABLE;
+ this.sslMode =
+ sslMode != null
+ ? (sslMode.isEmpty() ? SslMode.VERIFY_FULL : SslMode.from(sslMode))
+ : SslMode.DISABLE;
  this.enabledSslCipherSuites = enabledSslCipherSuites;
  this.sessionVariables = sessionVariables;
  this.tinyInt1isBit = tinyInt1isBit != null ? tinyInt1isBit : true;
@@ -493,16 +496,6 @@ private static void mapPropertiesToOption(Builder builder, Properties properties
  String.format(
  "Optional parameter %s must be Long, was '%s'", key, propertyValue));
  }
- } else if (field.getGenericType().equals(SslMode.class)) {
- if (propertyValue.isEmpty()) {
- field.set(builder, SslMode.VERIFY_FULL);
- } else {
- try {
- field.set(builder, SslMode.from(propertyValue));
- } catch (IllegalArgumentException i) {
- throw new SQLException(i.getMessage());
- }
- }
  }
  } catch (NoSuchFieldException nfe) {
  // keep unknown option:
@@ -1129,7 +1122,7 @@ public Builder credentialType(String credentialType) {
  return this;
  }
 
- public Builder sslMode(String string) {
+ public Builder sslMode(String sslMode) {
  this.sslMode = sslMode;
  return this;
  }

src/main/java/org/mariadb/jdbc/client/socket/PacketWriter.java

@@ -237,8 +237,9 @@ public void writeBytes(byte[] arr, int off, int len) throws IOException {
  if (mark != -1) {
  flushBufferStopAtMark();
  }
+ }
 
- } else {
+ if (len > buf.length - pos) {
  // not enough space in buf, will stream :
  // fill buf and flush until all data are snd
  int remainingLen = len;
@@ -641,20 +642,22 @@ private void growBuffer(int len) throws IOException {
  } else {
  newCapacity = maxPacketLength;
  }
+ if (len + pos > newCapacity) {
+ if (mark != -1) {
+ // buf is > 16M with mark.
+ // flush until mark, reset pos at beginning
+ flushBufferStopAtMark();
 
- if (mark != -1 && len + pos > newCapacity) {
- // buf is > 16M with mark.
- // flush until mark, reset pos at beginning
- flushBufferStopAtMark();
-
- if (len + pos <= bufLength) {
- return;
- }
+ if (len + pos <= bufLength) {
+ return;
+ }
 
- // need to keep all data, buf can grow more than maxPacketLength
- // grow buf if needed
- if (len + pos > newCapacity) {
- newCapacity = len + pos;
+ // need to keep all data, buf can grow more than maxPacketLength
+ // grow buf if needed
+ if (bufLength == maxPacketLength) return;
+ if (len + pos > newCapacity) {
+ newCapacity = Math.min(maxPacketLength, len + pos);
+ }
  }
  }
 
@@ -769,6 +772,10 @@ public boolean isMarked() {
  return mark != -1;
  }
 
+ public boolean hasFlushed() {
+ return sequence.get() != -1;
+ }
+
  /**
  * Flush to last mark.
  *

src/main/java/org/mariadb/jdbc/message/client/BulkExecutePacket.java

@@ -113,6 +113,22 @@ public int encode(PacketWriter writer, Context context, PrepareResultPacket newP
  param.encodeBinary(writer, context);
  }
  }
+
+ if (!writer.isMarked() && writer.hasFlushed()) {
+ // parameter were too big to fit in a MySQL packet
+ // need to finish the packet separately
+ writer.flush();
+ if (!paramIterator.hasNext()) {
+ break main_loop;
+ }
+ parameters = paramIterator.next();
+ // reset header type
+ for (int j = 0; j < parameterCount; j++) {
+ parameterHeaderType[j] = parameters.get(j);
+ }
+ break parameter_loop;
+ }
+
  writer.mark();
 
  if (writer.bufIsDataAfterMark()) {

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

@@ -23,14 +23,11 @@
 
 import static org.junit.jupiter.api.Assertions.*;
 
-import ch.qos.logback.classic.Level;
-import ch.qos.logback.classic.LoggerContext;
 import java.sql.*;
 import org.junit.jupiter.api.*;
 import org.mariadb.jdbc.Common;
 import org.mariadb.jdbc.Connection;
 import org.mariadb.jdbc.Statement;
-import org.slf4j.LoggerFactory;
 
 public class BatchTest extends Common {
 
@@ -178,20 +175,14 @@ public void largeBatch(Connection con) throws SQLException {
 
  @Test
  public void bulkPacketSplitMaxAllowedPacket() throws SQLException {
- LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory();
- ch.qos.logback.classic.Logger logger = loggerContext.getLogger("org.mariadb.jdbc");
- Level initialLvl = logger.getLevel();
- logger.setLevel(Level.TRACE);
- try {
- bulkPacketSplit(2, getMaxAllowedPacket() - 40);
- } finally {
- logger.setLevel(initialLvl);
- }
+ int maxAllowedPacket = getMaxAllowedPacket();
+ System.out.println("maxAllowedPacket:" + maxAllowedPacket);
+ bulkPacketSplit(2, maxAllowedPacket - 40);
  }
 
  @Test
  public void bulkPacketSplitMultiplePacket() throws SQLException {
- bulkPacketSplit(getMaxAllowedPacket() / 800, 1000);
+ bulkPacketSplit(4, getMaxAllowedPacket() / 3);
  }
 
  @Test

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

@@ -60,6 +60,7 @@ public void switchUser() throws SQLException {
  stmt.execute("CREATE USER 'dsUser'@'%'");
  stmt.execute("GRANT SELECT ON *.* TO 'dsUser'@'%' IDENTIFIED BY 'password'");
  }
+ stmt.execute("FLUSH PRIVILEGES");
 
  DataSource ds = new MariaDbDataSource(mDefUrl);
  try (Connection con1 = ds.getConnection()) {

src/test/java/org/mariadb/jdbc/unit/util/ConfigurationTest.java

@@ -583,9 +583,8 @@ public void builder() throws SQLException {
  .serverRsaPublicKeyFile("RSAPath")
  .allowPublicKeyRetrieval(true)
  .build();
- System.out.println(conf.toString());
  assertEquals(
- "jdbc:mariadb://address=(host=host1)(port=3305)(type=primary),address=(host=host2)(port=3307)(type=replica)/db?user=me&password=pwd&socketFactory=someSocketFactory&pipe=pipeName&localSocket=localSocket&localSocketAddress=localSocketAddress&credentialType=ENV&enabledSslCipherSuites=myCipher,cipher2&sessionVariables=blabla&tinyInt1isBit=false&yearIsDateType=false&timezone=UTC&connectionAttributes=bla=bla&useBulkStmts=false&autocommit=false&servicePrincipalName=SPN&tlsSocketType=TLStype&galeraAllowedState=A,B&enabledSslProtocolSuites=TLSv1.2&pinGlobalTxToPhysicalConnection=false&poolName=myPool&useReadAheadInput=false&cachePrepStmts=false&serverSslCert=mycertPath&serverRsaPublicKeyFile=RSAPath",
+ "jdbc:mariadb://address=(host=host1)(port=3305)(type=primary),address=(host=host2)(port=3307)(type=replica)/db?user=me&password=pwd&timezone=UTC&autocommit=false&pinGlobalTxToPhysicalConnection=false&socketFactory=someSocketFactory&pipe=pipeName&localSocket=localSocket&localSocketAddress=localSocketAddress&useReadAheadInput=false&tlsSocketType=TLStype&sslMode=REQUIRED&serverSslCert=mycertPath&enabledSslCipherSuites=myCipher,cipher2&enabledSslProtocolSuites=TLSv1.2&useBulkStmts=false&cachePrepStmts=false&credentialType=ENV&sessionVariables=blabla&connectionAttributes=bla=bla&servicePrincipalName=SPN&tinyInt1isBit=false&yearIsDateType=false&galeraAllowedState=A,B&poolName=myPool&serverRsaPublicKeyFile=RSAPath",
  conf.toString());
  }
 }