Do not send MySQL batches at once if pipelining is disabled (#1257)

* Do not send MySQL batches at once if pipelining is disabled

Fixes #1254

For example, if you send queries through ProxySQL, the batched operation will fail.

Also:
- make sure errors are reported correctly
- created a ProxySQL rule to be able to easily create more tests for it

Signed-off-by: Thomas Segismont <[email protected]>

* ExtendedBatchQueryCommandCodec: use Bitset instead of array of booleans

Signed-off-by: Thomas Segismont <[email protected]>

Signed-off-by: Thomas Segismont <[email protected]>

vertx-mysql-client/src/main/asciidoc/index.adoc

@@ -699,7 +699,6 @@ You can also manage the lifecycle of prepared statements manually by creating a
 
 There is time when you want to batch insert data into the database, you can use `PreparedQuery#executeBatch` which provides a simple API to handle this.
 Keep in mind that MySQL does not natively support batching protocol so the API is only a sugar by executing the prepared statement one after another, which means more network round trips are required comparing to inserting multiple rows by executing one prepared statement with a list of values.
-In order to gain best performance on the wire, we execute the batch in pipelining mode which means the execution request is sent before the response of previous request returns, if your server or proxy does not support this feature or you might not be interested in it, you can use the single execution API and compose them by yourself instead.
 
 === tricky DATE & TIME data types
 

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

@@ -20,81 +20,107 @@
 import io.vertx.sqlclient.impl.command.CommandResponse;
 import io.vertx.sqlclient.impl.command.ExtendedQueryCommand;
 
+import java.util.BitSet;
 import java.util.List;
 
-import static io.vertx.mysqlclient.impl.protocol.Packets.EnumCursorType.CURSOR_TYPE_NO_CURSOR;
+import static io.vertx.mysqlclient.impl.protocol.Packets.EnumCursorType.*;
 
 class ExtendedBatchQueryCommandCodec<R> extends ExtendedQueryCommandBaseCodec<R, ExtendedQueryCommand<R>> {
 
   private final List<TupleInternal> params;
-  private int batchIdx = 0;
+  private final BitSet bindingFailures;
+  private boolean pipeliningEnabled;
+  private int sent;
+  private int received;
 
   ExtendedBatchQueryCommandCodec(ExtendedQueryCommand<R> cmd) {
     super(cmd);
     params = cmd.paramsList();
+    bindingFailures = new BitSet(params.size());
   }
 
   @Override
   void encode(MySQLEncoder encoder) {
     super.encode(encoder);
-
     if (params.isEmpty() && statement.paramDesc.paramDefinitions().length > 0) {
       encoder.handleCommandResponse(CommandResponse.failure("Statement parameter is not set because of the empty batch param list"));
       return;
     }
+    pipeliningEnabled = encoder.socketConnection.pipeliningEnabled();
     encoder.socketConnection.suspendPipeline();
     doExecuteBatch();
-    // Close managed prepare statement
-    MySQLPreparedStatement ps = (MySQLPreparedStatement) this.cmd.ps;
-    if (ps.closeAfterUsage) {
-      sendCloseStatementCommand(ps);
-    }
   }
 
   @Override
   void handleErrorPacketPayload(ByteBuf payload) {
+    skipBindingFailures();
     MySQLException mySQLException = decodeErrorPacketPayload(payload);
-    reportError(batchIdx, mySQLException);
+    reportError(received++, mySQLException);
     // state needs to be reset
     commandHandlerState = CommandHandlerState.INIT;
-    batchIdx++;
+    if (received == params.size()) {
+      super.closePreparedStatement();
+      encoder.socketConnection.resumePipeline();
+      encoder.handleCommandResponse(CommandResponse.failure(failure));
+    } else {
+      doExecuteBatch();
+    }
+  }
+
+  private void skipBindingFailures() {
+    received = bindingFailures.nextClearBit(received);
   }
 
   @Override
   protected void handleSingleResultsetDecodingCompleted(int serverStatusFlags, int affectedRows, long lastInsertId) {
-    batchIdx++;
+    skipBindingFailures();
+    received++;
     super.handleSingleResultsetDecodingCompleted(serverStatusFlags, affectedRows, lastInsertId);
+    doExecuteBatch();
   }
 
   @Override
   protected boolean isDecodingCompleted(int serverStatusFlags) {
-    return super.isDecodingCompleted(serverStatusFlags) && batchIdx == params.size();
+    return super.isDecodingCompleted(serverStatusFlags) && received == params.size();
   }
 
   @Override
   protected void handleAllResultsetDecodingCompleted() {
     encoder.socketConnection.resumePipeline();
+    super.closePreparedStatement();
     super.handleAllResultsetDecodingCompleted();
   }
 
+  @Override
+  protected void closePreparedStatement() {
+    // Handled manually at the end of all executions
+  }
+
   private void doExecuteBatch() {
-    for (int i = 0; i < params.size(); i++) {
-      Tuple param = params.get(i);
+    while (sent < params.size()) {
+      Tuple param = params.get(sent);
       sequenceId = 0;
       // binding parameters
       String bindMsg = statement.bindParameters(param);
       if (bindMsg != null) {
-        reportError(i, new NoStackTraceThrowable(bindMsg));
+        bindingFailures.set(sent);
+        reportError(sent, new NoStackTraceThrowable(bindMsg));
+        sent++;
       } else {
         sendStatementExecuteCommand(statement, statement.sendTypesToServer(), param, CURSOR_TYPE_NO_CURSOR);
+        sent++;
+        if (!pipeliningEnabled) {
+          break;
+        }
       }
     }
   }
 
   private void reportError(int iteration, Throwable error) {
-    if (failure == null) {
-      failure = new MySQLBatchException();
+    MySQLBatchException batchException = (MySQLBatchException) failure;
+    if (batchException == null) {
+      failure = batchException = new MySQLBatchException();
     }
-    ((MySQLBatchException) failure).reportError(iteration, error);
+    batchException.reportError(iteration, error);
   }
 }

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

@@ -51,7 +51,7 @@ protected void handleAllResultsetDecodingCompleted() {
     super.handleAllResultsetDecodingCompleted();
   }
 
-  private void closePreparedStatement() {
+  protected void closePreparedStatement() {
     MySQLPreparedStatement ps = (MySQLPreparedStatement) this.cmd.ps;
     if (ps.closeAfterUsage) {
       sendCloseStatementCommand(ps);

vertx-mysql-client/src/test/java/io/vertx/mysqlclient/MySQLBatchInsertExceptionTest.java

@@ -0,0 +1,19 @@
+/*
+ * Copyright (c) 2011-2020 Contributors to the Eclipse Foundation
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
+ * which is available at https://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+ */
+
+package io.vertx.mysqlclient;
+
+import io.vertx.ext.unit.junit.VertxUnitRunner;
+import org.junit.runner.RunWith;
+
+@RunWith(VertxUnitRunner.class)
+public class MySQLBatchInsertExceptionTest extends MySQLBatchInsertExceptionTestBase {
+}

vertx-mysql-client/src/test/java/io/vertx/mysqlclient/MySQLPipeliningTest.java → vertx-mysql-client/src/test/java/io/vertx/mysqlclient/MySQLBatchInsertExceptionTestBase.java

@@ -13,40 +13,46 @@
 
 import io.vertx.core.Vertx;
 import io.vertx.ext.unit.TestContext;
-import io.vertx.ext.unit.junit.VertxUnitRunner;
-import io.vertx.sqlclient.*;
+import io.vertx.sqlclient.PoolOptions;
+import io.vertx.sqlclient.Row;
+import io.vertx.sqlclient.SqlClient;
+import io.vertx.sqlclient.Tuple;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
-import org.junit.runner.RunWith;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.stream.IntStream;
+
+import static java.util.stream.Collectors.*;
+
+public abstract class MySQLBatchInsertExceptionTestBase extends MySQLTestBase {
 
-@RunWith(VertxUnitRunner.class)
-public class MySQLPipeliningTest extends MySQLTestBase {
   Vertx vertx;
   MySQLConnectOptions options;
 
   @Before
   public void setup(TestContext ctx) {
     vertx = Vertx.vertx();
-    options = new MySQLConnectOptions(MySQLTestBase.options);
-    options.setPipeliningLimit(64);
+    options = createOptions();
     cleanTestTable(ctx);
   }
 
+  protected MySQLConnectOptions createOptions() {
+    return new MySQLConnectOptions(rule.options());
+  }
+
   @After
   public void teardown(TestContext ctx) {
     vertx.close(ctx.asyncAssertSuccess());
   }
 
   @Test
   public void testBatchInsertExceptionConn(TestContext ctx) {
-    MySQLConnection.connect(vertx, options)
-      .onComplete(ctx.asyncAssertSuccess(conn -> {
-        testBatchInsertException(ctx, conn);
-      }));
+    MySQLConnection.connect(vertx, options).onComplete(ctx.asyncAssertSuccess(conn -> {
+      testBatchInsertException(ctx, conn);
+    }));
   }
 
   @Test
@@ -56,30 +62,31 @@ public void testBatchInsertExceptionPool(TestContext ctx) {
   }
 
   private void testBatchInsertException(TestContext ctx, SqlClient client) {
+    int total = 50;
     List<Tuple> batchParams = new ArrayList<>();
-    for (int i = 0; i < 1000; i++) {
-      batchParams.add(Tuple.of(i, String.format("val-%d", i)));
+    for (int i = 0; i < total; i++) {
+      int val = (i & 1) == 1 ? i - 1 : i;
+      batchParams.add(Tuple.of(val, String.format("val-%d", val))); // primary key violation error occurs on odd numbers
     }
-    batchParams.add(501, Tuple.of(500, "error")); // primary key violation error occurs in the 501st iteration
 
     client.preparedQuery("INSERT INTO mutable(id, val) VALUES (?, ?)")
       .executeBatch(batchParams)
       .onComplete(ctx.asyncAssertFailure(error -> {
         ctx.assertEquals(MySQLBatchException.class, error.getClass());
         MySQLBatchException mySQLBatchException = (MySQLBatchException) error;
-        ctx.assertTrue(mySQLBatchException.getIterationError().containsKey(501));
+        ctx.assertEquals(IntStream.iterate(1, i -> i + 2).limit(total / 2).boxed().collect(toSet()), mySQLBatchException.getIterationError().keySet());
 
         // all the param will be executed
         client.query("SELECT id, val FROM mutable")
           .execute()
           .onComplete(ctx.asyncAssertSuccess(res2 -> {
-            ctx.assertEquals(1000, res2.size());
+            ctx.assertEquals(total / 2, res2.size());
             int i = 0;
             for (Row row : res2) {
               ctx.assertEquals(2, row.size());
               ctx.assertEquals(i, row.getInteger(0));
               ctx.assertEquals(String.format("val-%d", i), row.getString(1));
-              i++;
+              i += 2;
             }
             client.close();
           }));

vertx-mysql-client/src/test/java/io/vertx/mysqlclient/MySQLPipelinedBatchInsertExceptionTest.java

@@ -0,0 +1,23 @@
+/*
+ * Copyright (c) 2011-2020 Contributors to the Eclipse Foundation
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
+ * which is available at https://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+ */
+
+package io.vertx.mysqlclient;
+
+import io.vertx.ext.unit.junit.VertxUnitRunner;
+import org.junit.runner.RunWith;
+
+@RunWith(VertxUnitRunner.class)
+public class MySQLPipelinedBatchInsertExceptionTest extends MySQLBatchInsertExceptionTestBase {
+
+  protected MySQLConnectOptions createOptions() {
+    return super.createOptions().setPipeliningLimit(64);
+  }
+}

vertx-mysql-client/src/test/java/io/vertx/mysqlclient/ProxySQLBatchInsertExceptionTest.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright (c) 2011-2020 Contributors to the Eclipse Foundation
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
+ * which is available at https://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+ */
+
+package io.vertx.mysqlclient;
+
+import io.vertx.ext.unit.junit.VertxUnitRunner;
+import io.vertx.mysqlclient.junit.ProxySQLRule;
+import org.junit.ClassRule;
+import org.junit.runner.RunWith;
+
+@RunWith(VertxUnitRunner.class)
+public class ProxySQLBatchInsertExceptionTest extends MySQLBatchInsertExceptionTestBase {
+
+  @ClassRule
+  public static ProxySQLRule proxySql = new ProxySQLRule(rule);
+
+  @Override
+  protected MySQLConnectOptions createOptions() {
+    return proxySql.options(super.createOptions());
+  }
+}

vertx-mysql-client/src/test/java/io/vertx/mysqlclient/junit/ProxySQLRule.java

@@ -0,0 +1,83 @@
+package io.vertx.mysqlclient.junit;
+
+import io.vertx.mysqlclient.MySQLConnectOptions;
+import org.junit.rules.ExternalResource;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+
+import java.util.concurrent.TimeUnit;
+
+import static java.lang.String.*;
+
+public class ProxySQLRule extends ExternalResource {
+
+  private static final String MONITORING_USER = "proxysql";
+  private static final String MONITORING_USER_PASSWORD = "proxysql1234#";
+
+  private final MySQLRule mySQLRule;
+  private GenericContainer<?> proxySql;
+
+  public ProxySQLRule(MySQLRule mySQLRule) {
+    this.mySQLRule = mySQLRule;
+  }
+
+  @Override
+  protected void before() throws Throwable {
+    proxySql = new GenericContainer<>("proxysql/proxysql")
+      .withCreateContainerCmdModifier(createContainerCmd -> {
+        createContainerCmd.getHostConfig().withNetworkMode(mySQLRule.network());
+      })
+      .withExposedPorts(6032, 6033, 6070)
+      .waitingFor(Wait.forLogMessage(".*Latest ProxySQL version available.*\\n", 1));
+
+    proxySql.start();
+
+    execStatement("UPDATE global_variables SET variable_value='false' WHERE variable_name='admin-hash_passwords'", 10);
+    execStatement("LOAD ADMIN VARIABLES TO RUNTIME");
+
+    execStatement(format("UPDATE global_variables SET variable_value='%s' WHERE variable_name='mysql-monitor_username'", MONITORING_USER));
+    execStatement(format("UPDATE global_variables SET variable_value='%s' WHERE variable_name='mysql-monitor_password'", MONITORING_USER_PASSWORD));
+
+    execStatement("LOAD MYSQL VARIABLES TO RUNTIME");
+
+    execStatement(format("INSERT INTO mysql_servers(hostgroup_id, hostname, port) VALUES (0,'%s',3306)", mySQLRule.networkAlias()));
+    execStatement("LOAD MYSQL SERVERS TO RUNTIME");
+
+    execStatement(format("INSERT INTO mysql_users (username,password) VALUES ('%s','%s')", mySQLRule.options().getUser(), mySQLRule.options().getPassword()));
+    execStatement("LOAD MYSQL USERS TO RUNTIME");
+  }
+
+  private void execStatement(String statement) throws Exception {
+    execStatement(statement, 0);
+  }
+
+  private void execStatement(String statement, int retry) throws Exception {
+    RuntimeException failure;
+    for (int i = 0; ; i++) {
+      Container.ExecResult result = proxySql.execInContainer("mysql", "-u", "admin", "-p" + "admin", "-h", "127.0.0.1", "-P", "6032", "-e", statement);
+      if (result.getExitCode() == 0) {
+        return;
+      }
+      if (i >= retry && !result.getStderr().contains("ERROR 2002")) {
+        failure = new RuntimeException("Failed to execute statement: " + statement + "\n" + result.getStderr());
+        break;
+      }
+      TimeUnit.MILLISECONDS.sleep(200);
+    }
+    throw failure;
+  }
+
+  public MySQLConnectOptions options(MySQLConnectOptions other) {
+    return new MySQLConnectOptions(other)
+      .setHost(proxySql.getHost())
+      .setPort(proxySql.getMappedPort(6033));
+  }
+
+  @Override
+  protected void after() {
+    if (proxySql != null) {
+      proxySql.stop();
+    }
+  }
+}