MySQL cursor fetch uses wrong row descriptor after initial fetch (#1562) (#1564)

See #1525

MySQL sends column definitions in response to a prepare command.
But when executing the statement or fetching the first page of a cursor, it sends column definitions again.

The row decoder should not use the definitions given in response to prepare command because they may not be accurate.

The previous cursor implementation used them when fetching the second page.

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

Author: tsegismont

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

@@ -17,6 +17,7 @@
 package io.vertx.mysqlclient.impl.codec;
 
 import io.netty.buffer.ByteBuf;
+import io.vertx.mysqlclient.impl.MySQLRowDesc;
 import io.vertx.mysqlclient.impl.protocol.CommandType;
 import io.vertx.sqlclient.Tuple;
 import io.vertx.sqlclient.impl.command.CommandResponse;
@@ -31,7 +32,14 @@ class ExtendedQueryCommandCodec<R> extends ExtendedQueryCommandBaseCodec<R, Exte
     super(cmd);
     if (cmd.fetch() > 0 && statement.isCursorOpen) {
       // restore the state we need for decoding fetch response based on the prepared statement
-      columnDefinitions = statement.rowDesc.columnDefinitions();
+      columnDefinitions = statement.cursorRowDescriptor.columnDefinitions();
+    }
+  }
+
+  @Override
+  protected void handleRowDescriptorCreated(MySQLRowDesc mySQLRowDesc) {
+    if (cmd.fetch() > 0) {
+      statement.cursorRowDescriptor = mySQLRowDesc;
     }
   }
 
@@ -42,7 +50,7 @@ void encode(MySQLEncoder encoder) {
     if (statement.isCursorOpen) {
       if (decoder == null) {
         // restore the state we need for decoding if column definitions are not included in the fetch response
-        decoder = new RowResultDecoder<>(cmd.collector(), statement.rowDesc);
+        decoder = new RowResultDecoder<>(cmd.collector(), statement.cursorRowDescriptor);
       }
       sendStatementFetchCommand(statement.statementId, cmd.fetch());
     } else {

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

@@ -31,18 +31,17 @@ class MySQLPreparedStatement implements PreparedStatement {
   final long statementId;
   final String sql;
   final MySQLParamDesc paramDesc;
-  final MySQLRowDesc rowDesc;
   final boolean closeAfterUsage;
 
   private boolean sendTypesToServer;
   private final DataType[] bindingTypes;
 
   boolean isCursorOpen;
+  MySQLRowDesc cursorRowDescriptor;
 
-  MySQLPreparedStatement(String sql, long statementId, MySQLParamDesc paramDesc, MySQLRowDesc rowDesc, boolean closeAfterUsage) {
+  MySQLPreparedStatement(String sql, long statementId, MySQLParamDesc paramDesc, boolean closeAfterUsage) {
     this.statementId = statementId;
     this.paramDesc = paramDesc;
-    this.rowDesc = rowDesc;
     this.sql = sql;
     this.closeAfterUsage = closeAfterUsage;
 
@@ -58,7 +57,7 @@ public ParamDesc paramDesc() {
 
   @Override
   public RowDesc rowDesc() {
-    return rowDesc;
+    throw new UnsupportedOperationException("The client should use the column definitions provided by execute or fetch response instead of prepare response");
   }
 
   @Override

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

@@ -18,8 +18,6 @@
 
 import io.netty.buffer.ByteBuf;
 import io.vertx.mysqlclient.impl.MySQLParamDesc;
-import io.vertx.mysqlclient.impl.MySQLRowDesc;
-import io.vertx.mysqlclient.impl.datatype.DataFormat;
 import io.vertx.mysqlclient.impl.protocol.ColumnDefinition;
 import io.vertx.mysqlclient.impl.protocol.CommandType;
 import io.vertx.sqlclient.impl.PreparedStatement;
@@ -135,7 +133,6 @@ private void handleReadyForQuery() {
       cmd.sql(),
       this.statementId,
       new MySQLParamDesc(paramDescs),
-      MySQLRowDesc.create(columnDescs, DataFormat.BINARY),
       !cmd.isManaged())));
   }
 

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

@@ -94,9 +94,13 @@ protected void handleResultsetColumnDefinitions(ByteBuf payload) {
   protected void handleResultsetColumnDefinitionsDecodingCompleted() {
     commandHandlerState = CommandHandlerState.HANDLING_ROW_DATA_OR_END_PACKET;
     MySQLRowDesc mySQLRowDesc = MySQLRowDesc.create(columnDefinitions, format); // use the column definitions if provided by execute or fetch response instead of prepare response
+    handleRowDescriptorCreated(mySQLRowDesc);
     decoder = new RowResultDecoder<>(cmd.collector(), mySQLRowDesc);
   }
 
+  protected void handleRowDescriptorCreated(MySQLRowDesc mySQLRowDesc) {
+  }
+
   protected void handleRows(ByteBuf payload, int payloadLength) {
   /*
     Resultset row can begin with 0xfe byte (when using text protocol with a field length > 0xffffff)

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

@@ -29,6 +29,7 @@ void encode(MySQLEncoder encoder) {
     statement.cleanBindings();
 
     statement.isCursorOpen = false;
+    statement.cursorRowDescriptor = null;
     sendStatementResetCommand(statement.statementId);
   }
 

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

@@ -14,11 +14,11 @@
 import io.vertx.core.Vertx;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.file.FileSystem;
+import io.vertx.ext.unit.Async;
 import io.vertx.ext.unit.TestContext;
 import io.vertx.ext.unit.junit.VertxUnitRunner;
 import io.vertx.sqlclient.*;
 import org.junit.After;
-import org.junit.Assume;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -29,6 +29,8 @@
 import java.util.List;
 import java.util.concurrent.ThreadLocalRandom;
 
+import static org.junit.Assume.assumeFalse;
+
 @RunWith(VertxUnitRunner.class)
 public class MySQLQueryTest extends MySQLTestBase {
 
@@ -221,7 +223,7 @@ public void testDecodePacketSizeMoreThan16MB(TestContext ctx) {
 
   @Test
   public void testEncodePacketSizeMoreThan16MB(TestContext ctx) {
-    Assume.assumeFalse(rule.isUsingMySQL5_6());
+    assumeFalse(rule.isUsingMySQL5_6());
     int dataSize = 20 * 1024 * 1024; // 20MB payload
     byte[] data = new byte[dataSize];
     ThreadLocalRandom.current().nextBytes(data);
@@ -366,4 +368,22 @@ public void testLocalInfileRequestInPackets(TestContext ctx) {
       }));
     }));
   }
+
+  @Test
+  public void testColumnDefinitionChangeWithCursor(TestContext ctx) {
+    assumeFalse("Query syntax not supported on MySQL 5", rule.isUsingMySQL5());
+    Async rows = ctx.async(100);
+    Async completion = ctx.async();
+    MySQLConnection.connect(vertx, options).onComplete(ctx.asyncAssertSuccess(conn -> {
+      conn
+        .prepare("SELECT row_number() over () as \"Number\", case when 1 then 1 else 0 end as \"Case\" FROM mysql.help_relation limit 0,100")
+        .onComplete(ctx.asyncAssertSuccess(ps -> {
+          // Make sure to fetch from the database twice
+          RowStream<Row> stream = ps.createStream(50);
+          stream.exceptionHandler(ctx::fail);
+          stream.endHandler(v -> completion.complete());
+          stream.handler(row -> rows.countDown());
+        }));
+    }));
+  }
 }

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

@@ -142,6 +142,10 @@ public boolean isUsingMariaDB() {
     return databaseServerInfo.getDatabaseType() == DatabaseType.MariaDB;
   }
 
+  public boolean isUsingMySQL5() {
+    return databaseServerInfo.databaseType == DatabaseType.MySQL && databaseServerInfo.dockerImageTag.startsWith("5.");
+  }
+
   public boolean isUsingMySQL5_6() {
     return databaseServerInfo == DatabaseServerInfo.MySQL_V5_6;
   }

vertx-mysql-client/src/test/resources/init.sql

@@ -1,3 +1,8 @@
+#allow reading mysql schema
+GRANT
+SELECT
+ON mysql.* TO 'mysql';
+
 # testing change schema
 CREATE DATABASE emptyschema;
 GRANT ALL ON emptyschema.* TO 'mysql'@'%';