/*
 * Copyright (c) 2015, 2023, Oracle and/or its affiliates.
 *
 * This program is free software; you can redistribute it and/or modify it under
 * the terms of the GNU General Public License, version 2.0, as published by the
 * Free Software Foundation.
 *
 * This program is also distributed with certain software (including but not
 * limited to OpenSSL) that is licensed under separate terms, as designated in a
 * particular file or component or in included license documentation. The
 * authors of MySQL hereby grant you an additional permission to link the
 * program and your derivative works with the separately licensed software that
 * they have included with MySQL.
 *
 * Without limiting anything contained in the foregoing, this file, which is
 * part of MySQL Connector/J, is also subject to the Universal FOSS Exception,
 * version 1.0, a copy of which can be found at
 * http://oss.oracle.com/licenses/universal-foss-exception.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
 * FOR A PARTICULAR PURPOSE. See the GNU General Public License, version 2.0,
 * for more details.
 *
 * You should have received a copy of the GNU General Public License along with
 * this program; if not, write to the Free Software Foundation, Inc.,
 * 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
 */

package testsuite.x.devapi;

import static org.hamcrest.CoreMatchers.hasItems;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assumptions.assumeTrue;

import java.lang.reflect.Field;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Function;
import java.util.stream.Collectors;

import org.junit.jupiter.api.Test;

import com.mysql.cj.CoreSession;
import com.mysql.cj.ServerVersion;
import com.mysql.cj.conf.PropertyKey;
import com.mysql.cj.exceptions.DataConversionException;
import com.mysql.cj.protocol.x.XProtocol;
import com.mysql.cj.protocol.x.XProtocolError;
import com.mysql.cj.xdevapi.Column;
import com.mysql.cj.xdevapi.Row;
import com.mysql.cj.xdevapi.RowResult;
import com.mysql.cj.xdevapi.SelectStatement;
import com.mysql.cj.xdevapi.Session;
import com.mysql.cj.xdevapi.SessionFactory;
import com.mysql.cj.xdevapi.SessionImpl;
import com.mysql.cj.xdevapi.SqlResult;
import com.mysql.cj.xdevapi.Statement;
import com.mysql.cj.xdevapi.Table;
import com.mysql.cj.xdevapi.Type;

/**
 * @todo
 */
public class TableSelectTest extends BaseTableTestCase {

    @Test
    public void basicQuery() {
        try {
            sqlUpdate("drop table if exists basicQuery");
            sqlUpdate("create table basicQuery (_id varchar(32), name varchar(20), birthday date, age int)");
            sqlUpdate("insert into basicQuery values ('some long UUID', 'Sakila', '2000-05-27', 14)");
            Table table = this.schema.getTable("basicQuery");
            Map<String, Object> params = new HashMap<>();
            params.put("name", "Saki%");
            params.put("age", 20);
            RowResult rows = table.select("birthday, `_id`, name").where("name like :name AND age < :age").bind(params).execute();

            // verify metadata
            List<String> columnNames = rows.getColumnNames();
            assertEquals("birthday", columnNames.get(0));
            assertEquals("_id", columnNames.get(1));
            assertEquals("name", columnNames.get(2));

            Row row = rows.next();
            assertEquals("2000-05-27", row.getString(0));
            assertEquals("2000-05-27", row.getString("birthday"));
            assertEquals("Sakila", row.getString(2));
            assertEquals("Sakila", row.getString("name"));
            assertEquals("some long UUID", row.getString(1));
            assertEquals("some long UUID", row.getString("_id"));

            // select with multiple projection params
            rows = table.select("`_id`", "name", "birthday").where("name like :name AND age < :age").bind(params).execute();
            // verify metadata
            columnNames = rows.getColumnNames();
            assertEquals("_id", columnNames.get(0));
            assertEquals("name", columnNames.get(1));
            assertEquals("birthday", columnNames.get(2));
        } finally {
            sqlUpdate("drop table if exists basicQuery");
        }
    }

    @Test
    public void testComplexQuery() {
        try {
            sqlUpdate("drop table if exists complexQuery");
            sqlUpdate("create table complexQuery (name varchar(32), age int, something int)");
            sqlUpdate("insert into complexQuery values ('Mamie', 11, 0)");
            sqlUpdate("insert into complexQuery values ('Eulalia', 11, 0)");
            sqlUpdate("insert into complexQuery values ('Polly', 12, 0)");
            sqlUpdate("insert into complexQuery values ('Rufus', 12, 0)");
            sqlUpdate("insert into complexQuery values ('Cassidy', 13, 0)");
            sqlUpdate("insert into complexQuery values ('Olympia', 14, 0)");
            sqlUpdate("insert into complexQuery values ('Lev', 14, 0)");
            sqlUpdate("insert into complexQuery values ('Tierney', 15, 0)");
            sqlUpdate("insert into complexQuery values ('Octavia', 15, 0)");
            sqlUpdate("insert into complexQuery values ('Vesper', 16, 0)");
            sqlUpdate("insert into complexQuery values ('Caspian', 17, 0)");
            sqlUpdate("insert into complexQuery values ('Romy', 17, 0)");
            Table table = this.schema.getTable("complexQuery");
            // Result:
            // age_group | cnt
            // 11        | 2   <-- filtered out by where
            // 12        | 2   <-- filtered out by limit
            // 13        | 1   <-- filtered out by having
            // 14        | 2   * second row in result
            // 15        | 2   * first row in result
            // 16        | 1   <-- filtered out by having
            // 17        | 2   <-- filtered out by offset
            SelectStatement stmt = table.select("age as age_group, count(name) as cnt, something");
            stmt.where("age > 11 and 1 < 2 and 40 between 30 and 900");
            stmt.groupBy("something", "age_group");
            stmt.having("cnt > 1");
            stmt.orderBy("age_group desc");
            RowResult rows = stmt.limit(2).offset(1).execute();
            Row row = rows.next();
            assertEquals(15, row.getInt(0));
            assertEquals(2, row.getInt(1));
            assertEquals(2, row.getByte(1));
            assertEquals(2, row.getLong(1));
            assertEquals(new BigDecimal("2"), row.getBigDecimal(1));
            assertEquals(true, row.getBoolean(1));
            row = rows.next();
            assertEquals(14, row.getInt(0));
            assertEquals(2, row.getInt(1));
            assertFalse(rows.hasNext());
        } finally {
            sqlUpdate("drop table if exists complexQuery");
        }
    }

    @Test
    public void allColumns() {
        try {
            sqlUpdate("drop table if exists allColumns");
            sqlUpdate("create table allColumns (x int, y int, z int)");
            sqlUpdate("insert into allColumns values (1,2,3)");
            Table table = this.schema.getTable("allColumns");
            // * must come first, as with SQL
            SelectStatement stmt = table.select("*, 42 as a_number, '43' as a_string");
            Row row = stmt.execute().next();
            assertEquals(42, row.getInt("a_number"));
            assertEquals(1, row.getInt("x"));
            assertEquals(2, row.getInt("y"));
            assertEquals(3, row.getInt("z"));
            assertEquals("43", row.getString("a_string"));
        } finally {
            sqlUpdate("drop table if exists allColumns");
        }
    }

    @Test
    public void countAllColumns() {
        try {
            sqlUpdate("drop table if exists countAllColumns");
            sqlUpdate("create table countAllColumns(x int, y int)");
            sqlUpdate("insert into countAllColumns values (1,1), (2,2), (3,3), (4,4)");
            Table table = this.schema.getTable("countAllColumns");
            Row row = table.select("count(*) + 10").execute().next();
            assertEquals(14, row.getInt(0));
        } finally {
            sqlUpdate("drop table if exists countAllColumns");
        }
    }

    /**
     * Tests fix for Bug#22931433, GETTING VALUE OF BIT COLUMN RESULTS IN EXCEPTION.
     */
    @Test
    public void testBug22931433() {
        sqlUpdate("drop table if exists testBug22931433");
        sqlUpdate(
                "create table testBug22931433(c1 bit(8), c2 bit(16), c3 bit(24), c4 bit(32), c5 bit(40), c6 bit(48), c7 bit(56), c8 bit(64), cb1 bit(1), cb2 bit(64))");

        Table table = this.schema.getTable("testBug22931433");
        table.insert("c1", "c2", "c3", "c4", "c5", "c6", "c7", "c8", "cb1", "cb2")
                .values("a", "ba", "cba", "dcba", "edcba", "fedcba", "gfedcba", "hgfedcba", 0x01, -1).execute();
        table.insert("c1", "c2", "c3", "c4", "c5", "c6", "c7", "c8", "cb1", "cb2")
                .values(0xcc, 0xcccc, 0xcccccc, 0xccccccccL, 0xccccccccccL, 0xccccccccccccL, 0xccccccccccccccL, 0xccccccccccccccccL, 0x00, -2).execute();

        String testUrl = this.baseUrl + (this.baseUrl.contains("?") ? "&" : "?") + makeParam(PropertyKey.jdbcCompliantTruncation, "false", true);
        Session s1 = this.fact.getSession(testUrl);
        table = s1.getDefaultSchema().getTable("testBug22931433");

        RowResult rows = table.select("c1, c2, c3, c4, c5, c6, c7, c8, cb1, cb2").execute();

        Row row = rows.next();

        assertEquals('a', row.getByte("c1"));
        assertEquals('a', row.getByte("c2"));
        assertEquals('a', row.getByte("c3"));
        assertEquals('a', row.getByte("c4"));
        assertEquals('a', row.getByte("c5"));
        assertEquals('a', row.getByte("c6"));
        assertEquals('a', row.getByte("c7"));
        assertEquals('a', row.getByte("c8"));

        assertEquals(97, row.getInt("c1"));
        assertEquals(25185, row.getInt("c2"));
        assertEquals(6513249, row.getInt("c3"));
        assertEquals(1684234849, row.getInt("c4"));
        assertEquals(1684234849, row.getInt("c5")); // truncated to 4 bytes
        assertEquals(1684234849, row.getInt("c6")); // truncated to 4 bytes
        assertEquals(1684234849, row.getInt("c7")); // truncated to 4 bytes
        assertEquals(1684234849, row.getInt("c8")); // truncated to 4 bytes

        assertEquals(97, row.getLong("c1"));
        assertEquals(25185, row.getLong("c2"));
        assertEquals(6513249, row.getLong("c3"));
        assertEquals(1684234849, row.getLong("c4"));
        assertEquals(435475931745L, row.getLong("c5"));
        assertEquals(112585661964897L, row.getLong("c6"));
        assertEquals(29104508263162465L, row.getLong("c7"));
        assertEquals(7523094288207667809L, row.getLong("c8"));

        assertEquals(BigDecimal.valueOf(97), row.getBigDecimal("c1"));
        assertEquals(BigDecimal.valueOf(25185), row.getBigDecimal("c2"));
        assertEquals(BigDecimal.valueOf(6513249), row.getBigDecimal("c3"));
        assertEquals(BigDecimal.valueOf(1684234849), row.getBigDecimal("c4"));
        assertEquals(BigDecimal.valueOf(435475931745L), row.getBigDecimal("c5"));
        assertEquals(BigDecimal.valueOf(112585661964897L), row.getBigDecimal("c6"));
        assertEquals(BigDecimal.valueOf(29104508263162465L), row.getBigDecimal("c7"));
        assertEquals(BigDecimal.valueOf(7523094288207667809L), row.getBigDecimal("c8"));

        assertEquals(Double.valueOf(97), Double.valueOf(row.getDouble("c1")));
        assertEquals(Double.valueOf(25185), Double.valueOf(row.getDouble("c2")));
        assertEquals(Double.valueOf(6513249), Double.valueOf(row.getDouble("c3")));
        assertEquals(Double.valueOf(1684234849), Double.valueOf(row.getDouble("c4")));
        assertEquals(Double.valueOf(435475931745L), Double.valueOf(row.getDouble("c5")));
        assertEquals(Double.valueOf(112585661964897L), Double.valueOf(row.getDouble("c6")));
        assertEquals(Double.valueOf(29104508263162465L), Double.valueOf(row.getDouble("c7")));
        assertEquals(Double.valueOf(7523094288207667809L), Double.valueOf(row.getDouble("c8")));

        assertEquals(true, row.getBoolean("c1"));
        assertEquals(true, row.getBoolean("cb1"));
        assertEquals(true, row.getBoolean("cb2"));

        assertEquals(BigDecimal.valueOf(97).toString(), row.getString("c1"));
        assertEquals(BigDecimal.valueOf(25185).toString(), row.getString("c2"));
        assertEquals(BigDecimal.valueOf(6513249).toString(), row.getString("c3"));
        assertEquals(BigDecimal.valueOf(1684234849).toString(), row.getString("c4"));
        assertEquals(BigDecimal.valueOf(435475931745L).toString(), row.getString("c5"));
        assertEquals(BigDecimal.valueOf(112585661964897L).toString(), row.getString("c6"));
        assertEquals(BigDecimal.valueOf(29104508263162465L).toString(), row.getString("c7"));
        assertEquals(BigDecimal.valueOf(7523094288207667809L).toString(), row.getString("c8"));

        assertThrows(DataConversionException.class, "Unsupported conversion from BIT to java.sql.Date", () -> {
            row.getDate("c1");
            return null;
        });

        assertThrows(DataConversionException.class, "Unsupported conversion from BIT to com.mysql.cj.xdevapi.DbDoc", () -> {
            row.getDbDoc("c1");
            return null;
        });

        assertThrows(DataConversionException.class, "Unsupported conversion from BIT to java.sql.Time", () -> {
            row.getTime("c1");
            return null;
        });

        assertThrows(DataConversionException.class, "Unsupported conversion from BIT to java.sql.Timestamp", () -> {
            row.getTimestamp("c1");
            return null;
        });

        // test negative values
        Row row2 = rows.next();

        assertEquals(-52, row2.getByte("c1"));
        assertEquals(-52, row2.getByte("c2"));
        assertEquals(-52, row2.getByte("c3"));
        assertEquals(-52, row2.getByte("c4"));
        assertEquals(-52, row2.getByte("c5"));
        assertEquals(-52, row2.getByte("c6"));
        assertEquals(-52, row2.getByte("c7"));
        assertEquals(-52, row2.getByte("c8"));

        assertEquals(204, row2.getInt("c1"));
        assertEquals(52428, row2.getInt("c2"));
        assertEquals(13421772, row2.getInt("c3"));
        assertEquals(-858993460, row2.getInt("c4"));
        assertEquals(-858993460, row2.getInt("c5")); // truncated to 4 bytes
        assertEquals(-858993460, row2.getInt("c6")); // truncated to 4 bytes
        assertEquals(-858993460, row2.getInt("c7")); // truncated to 4 bytes
        assertEquals(-858993460, row2.getInt("c8")); // truncated to 4 bytes

        assertEquals(204, row2.getLong("c1"));
        assertEquals(52428, row2.getLong("c2"));
        assertEquals(13421772, row2.getLong("c3"));
        assertEquals(3435973836L, row2.getLong("c4"));
        assertEquals(879609302220L, row2.getLong("c5"));
        assertEquals(225179981368524L, row2.getLong("c6"));
        assertEquals(57646075230342348L, row2.getLong("c7"));
        assertEquals(-3689348814741910324L, row2.getLong("c8"));

        assertEquals(BigDecimal.valueOf(204), row2.getBigDecimal("c1"));
        assertEquals(BigDecimal.valueOf(52428), row2.getBigDecimal("c2"));
        assertEquals(BigDecimal.valueOf(13421772), row2.getBigDecimal("c3"));
        assertEquals(BigDecimal.valueOf(3435973836L), row2.getBigDecimal("c4"));
        assertEquals(BigDecimal.valueOf(879609302220L), row2.getBigDecimal("c5"));
        assertEquals(BigDecimal.valueOf(225179981368524L), row2.getBigDecimal("c6"));
        assertEquals(BigDecimal.valueOf(57646075230342348L), row2.getBigDecimal("c7"));
        assertEquals(new BigDecimal(new BigInteger("14757395258967641292")), row2.getBigDecimal("c8"));

        assertEquals(Double.valueOf(204), Double.valueOf(row2.getDouble("c1")));
        assertEquals(Double.valueOf(52428), Double.valueOf(row2.getDouble("c2")));
        assertEquals(Double.valueOf(13421772), Double.valueOf(row2.getDouble("c3")));
        assertEquals(Double.valueOf(3435973836L), Double.valueOf(row2.getDouble("c4")));
        assertEquals(Double.valueOf(879609302220L), Double.valueOf(row2.getDouble("c5")));
        assertEquals(Double.valueOf(225179981368524L), Double.valueOf(row2.getDouble("c6")));
        assertEquals(Double.valueOf(57646075230342348L), Double.valueOf(row2.getDouble("c7")));
        assertEquals(Double.valueOf(new BigInteger("14757395258967641292").doubleValue()), Double.valueOf(row2.getDouble("c8")));

        assertEquals(false, row2.getBoolean("c8"));
        assertEquals(false, row2.getBoolean("cb1"));
        assertEquals(false, row2.getBoolean("cb2"));

        sqlUpdate("drop table if exists testBug22931433");
        s1.close();
    }

    @Test
    public void basicViewQuery() {
        try {
            sqlUpdate("drop table if exists basicTable1");
            sqlUpdate("drop table if exists basicTable2");
            sqlUpdate("drop view if exists basicView");

            sqlUpdate("create table basicTable1 (_id varchar(32), name varchar(20))");
            sqlUpdate("create table basicTable2 (_id varchar(32), birthday date, age int)");
            sqlUpdate(
                    "create view basicView as select basicTable1._id, name, birthday, age from basicTable1 join basicTable2 on basicTable1._id=basicTable2._id");

            sqlUpdate("insert into basicTable1 values ('some long UUID', 'Sakila')");
            sqlUpdate("insert into basicTable2 values ('some long UUID', '2000-05-27', 14)");

            Table view = this.schema.getTable("basicView");
            Map<String, Object> params = new HashMap<>();
            params.put("name", "Saki%");
            params.put("age", 20);
            RowResult rows = view.select("birthday, `_id`, name").where("name like :name AND age < :age").bind(params).execute();

            // verify metadata
            List<String> columnNames = rows.getColumnNames();
            assertEquals("birthday", columnNames.get(0));
            assertEquals("_id", columnNames.get(1));
            assertEquals("name", columnNames.get(2));

            Row row = rows.next();
            assertEquals("2000-05-27", row.getString(0));
            assertEquals("2000-05-27", row.getString("birthday"));
            assertEquals("Sakila", row.getString(2));
            assertEquals("Sakila", row.getString("name"));
            assertEquals("some long UUID", row.getString(1));
            assertEquals("some long UUID", row.getString("_id"));
        } finally {
            sqlUpdate("drop table if exists basicTable1");
            sqlUpdate("drop table if exists basicTable2");
            sqlUpdate("drop view if exists basicView");
        }
    }

    @Test
    public void testOrderBy() {
        try {
            sqlUpdate("drop table if exists testOrderBy");
            sqlUpdate("create table testOrderBy (_id int, x int, y int)");
            sqlUpdate("insert into testOrderBy values (2,20,21), (1,20,22), (4,10,40), (3,10,50)");
            Table table = this.schema.getTable("testOrderBy");

            RowResult rows = table.select("_id").orderBy("x desc, y desc").execute();
            int i = 1;
            while (rows.hasNext()) {
                assertEquals(i++, rows.next().getInt("_id"));
            }
            assertEquals(5, i);

            // multiple SortExprStr
            rows = table.select("_id").orderBy("x desc", "y desc").execute();
            i = 1;
            while (rows.hasNext()) {
                assertEquals(i++, rows.next().getInt("_id"));
            }
            assertEquals(5, i);
        } finally {
            sqlUpdate("drop table if exists testOrderBy");
        }
    }

    @Test
    public void testBug22988922() {
        sqlUpdate("drop table if exists testBug22988922");
        sqlUpdate("create table testBug22988922 (g point,l longblob,t longtext)");

        Table table = this.schema.getTable("testBug22988922");

        RowResult rows = table.select("*").execute();
        List<Column> metadata = rows.getColumns();
        // assertEquals(4294967295L, metadata.get(0).getLength()); // irrelevant, we shouldn't expect any concrete value
        assertEquals(4294967295L, metadata.get(1).getLength());
        assertEquals(4294967295L, metadata.get(2).getLength());

        sqlUpdate("drop table if exists testBug22988922");
    }

    /**
     * Tests fix for BUG#22931277, COLUMN.GETTYPE() RETURNS ERROR FOR VALID DATATYPES.
     */
    @Test
    public void testBug22931277() {
        try {
            sqlUpdate("drop table if exists testBug22931277");
            sqlUpdate("create table testBug22931277 (j year,k datetime(3))");
            Table table = this.schema.getTable("testBug22931277");

            table = this.schema.getTable("testBug22931277");
            table.insert("j", "k").values(2000, "2016-03-15 12:13:14").execute();

            RowResult rows = table.select("1,j,k").execute();
            List<Column> metadata = rows.getColumns();

            Column myCol = metadata.get(0);
            assertEquals(Type.TINYINT, myCol.getType());

            myCol = metadata.get(1);
            assertEquals(Type.SMALLINT, myCol.getType());

            myCol = metadata.get(2);
            assertEquals(Type.DATETIME, myCol.getType());
        } finally {
            sqlUpdate("drop table if exists testBug22931277");
        }
    }

    @Test
    public void testTableRowLocks() throws Exception {
        assumeTrue(mysqlVersionMeetsMinimum(ServerVersion.parseVersion("8.0.3")), "MySQL 8.0.3+ is required to run this test.");

        sqlUpdate("drop table if exists testTableRowLocks");
        sqlUpdate("create table testTableRowLocks (_id varchar(32), a varchar(20))");
        sqlUpdate("CREATE UNIQUE INDEX myIndex ON testTableRowLocks (_id)"); // index is required to enable row locking
        sqlUpdate("insert into testTableRowLocks values ('1', '1')");
        sqlUpdate("insert into testTableRowLocks values ('2', '1')");
        sqlUpdate("insert into testTableRowLocks values ('3', '1')");

        Session session1 = null;
        Session session2 = null;

        try {
            session1 = new SessionFactory().getSession(this.testProperties);
            Table table1 = session1.getDefaultSchema().getTable("testTableRowLocks");
            session2 = new SessionFactory().getSession(this.testProperties);
            Table table2 = session2.getDefaultSchema().getTable("testTableRowLocks");

            // test1: Shared Lock
            session1.startTransaction();
            table1.select("_id").where("_id = '1'").lockShared().execute();

            session2.startTransaction();
            table2.select("_id").where("_id = '2'").lockShared().execute(); // should return immediately

            CompletableFuture<RowResult> res1 = table2.select("_id").where("_id = '1'").lockShared().executeAsync(); // should return immediately
            res1.get(5, TimeUnit.SECONDS);
            assertTrue(res1.isDone());

            session1.rollback();
            session2.rollback();

            // test2: Shared Lock after Exclusive
            session1.startTransaction();
            table1.select("_id").where("_id = '1'").lockExclusive().execute();

            session2.startTransaction();
            table2.select("_id").where("_id = '2'").lockShared().execute(); // should return immediately
            CompletableFuture<RowResult> res2 = table2.select("_id").where("_id = '1'").lockShared().executeAsync(); // session2 blocks
            assertThrows(TimeoutException.class, () -> {
                res2.get(5, TimeUnit.SECONDS);
                return null;
            });

            session1.rollback(); // session2 should unblock now
            res2.get(5, TimeUnit.SECONDS);
            assertTrue(res2.isDone());
            session2.rollback();

            // test3: Exclusive after Shared
            session1.startTransaction();
            table1.select("_id").where("_id = '1'").lockShared().execute();
            table1.select("_id").where("_id = '3'").lockShared().execute();

            session2.startTransaction();
            table2.select("_id").where("_id = '2'").lockExclusive().execute(); // should return immediately
            table2.select("_id").where("_id = '3'").lockShared().execute(); // should return immediately
            CompletableFuture<RowResult> res3 = table2.select("_id").where("_id = '1'").lockExclusive().executeAsync(); // session2 blocks
            assertThrows(TimeoutException.class, () -> {
                res3.get(5, TimeUnit.SECONDS);
                return null;
            });

            session1.rollback(); // session2 should unblock now
            res3.get(5, TimeUnit.SECONDS);
            assertTrue(res3.isDone());
            session2.rollback();

            // test4: Exclusive after Exclusive
            session1.startTransaction();
            table1.select("_id").where("_id = '1'").lockExclusive().execute();

            session2.startTransaction();
            table2.select("_id").where("_id = '2'").lockExclusive().execute(); // should return immediately
            CompletableFuture<RowResult> res4 = table2.select("_id").where("_id = '1'").lockExclusive().executeAsync(); // session2 blocks
            assertThrows(TimeoutException.class, () -> {
                res4.get(5, TimeUnit.SECONDS);
                return null;
            });

            session1.rollback(); // session2 should unblock now
            res4.get(5, TimeUnit.SECONDS);
            assertTrue(res4.isDone());
            session2.rollback();

        } finally {
            if (session1 != null) {
                session1.close();
            }
            if (session2 != null) {
                session2.close();
            }
            sqlUpdate("drop table if exists testTableRowLocks");
        }
    }

    @Test
    public void testTableRowLockOptions() throws Exception {
        assumeTrue(mysqlVersionMeetsMinimum(ServerVersion.parseVersion("8.0.5")), "MySQL 8.0.5+ is required to run this test.");

        Function<RowResult, List<String>> asStringList = rr -> rr.fetchAll().stream().map(r -> r.getString(0)).collect(Collectors.toList());

        sqlUpdate("DROP TABLE IF EXISTS testTableRowLockOptions");
        sqlUpdate("CREATE TABLE testTableRowLockOptions (_id VARCHAR(32), a VARCHAR(20))");
        sqlUpdate("CREATE UNIQUE INDEX myIndex ON testTableRowLockOptions (_id)"); // index is required to enable row locking
        sqlUpdate("INSERT INTO testTableRowLockOptions VALUES ('1', '1'), ('2', '1'), ('3', '1')");

        Session session1 = null;
        Session session2 = null;

        try {
            session1 = new SessionFactory().getSession(this.testProperties);
            Table table1 = session1.getDefaultSchema().getTable("testTableRowLockOptions");
            session2 = new SessionFactory().getSession(this.testProperties);
            Table table2 = session2.getDefaultSchema().getTable("testTableRowLockOptions");
            RowResult res;
            CompletableFuture<RowResult> futRes;

            /*
             * 1. Shared Lock in both sessions.
             */

            // session2.lockShared() returns data immediately.
            session1.startTransaction();
            table1.select("_id").where("_id = '1'").lockShared().execute();

            session2.startTransaction();
            res = table2.select("_id").where("_id < '3'").lockShared().execute();
            assertEquals(2, asStringList.apply(res).size());
            assertThat(asStringList.apply(res), hasItems("1", "2"));
            session2.rollback();

            session2.startTransaction();
            futRes = table2.select("_id").where("_id < '3'").lockShared().executeAsync();
            res = futRes.get(3, TimeUnit.SECONDS);
            assertTrue(futRes.isDone());
            assertEquals(2, asStringList.apply(res).size());
            assertThat(asStringList.apply(res), hasItems("1", "2"));
            session2.rollback();

            session1.rollback();

            // session2.lockShared(NOWAIT) returns data immediately.
            session1.startTransaction();
            table1.select("_id").where("_id = '1'").lockShared().execute();

            session2.startTransaction();
            res = table2.select("_id").where("_id < '3'").lockShared(Statement.LockContention.NOWAIT).execute();
            assertEquals(2, asStringList.apply(res).size());
            assertThat(asStringList.apply(res), hasItems("1", "2"));
            session2.rollback();

            session2.startTransaction();
            futRes = table2.select("_id").where("_id < '3'").lockShared(Statement.LockContention.NOWAIT).executeAsync();
            res = futRes.get(3, TimeUnit.SECONDS);
            assertTrue(futRes.isDone());
            assertEquals(2, asStringList.apply(res).size());
            assertThat(asStringList.apply(res), hasItems("1", "2"));
            session2.rollback();

            session1.rollback();

            // session2.lockShared(SKIP_LOCK) returns data immediately.
            session1.startTransaction();
            table1.select("_id").where("_id = '1'").lockShared().execute();

            session2.startTransaction();
            res = table2.select("_id").where("_id < '3'").lockShared(Statement.LockContention.SKIP_LOCKED).execute();
            assertEquals(2, asStringList.apply(res).size());
            assertThat(asStringList.apply(res), hasItems("1", "2"));
            session2.rollback();

            session2.startTransaction();
            futRes = table2.select("_id").where("_id < '3'").lockShared(Statement.LockContention.SKIP_LOCKED).executeAsync();
            res = futRes.get(3, TimeUnit.SECONDS);
            assertTrue(futRes.isDone());
            assertEquals(2, asStringList.apply(res).size());
            assertThat(asStringList.apply(res), hasItems("1", "2"));
            session2.rollback();

            session1.rollback();

            /*
             * 2. Shared Lock in first session and exclusive lock in second.
             */

            // session2.lockExclusive() blocks until session1 ends.
            session1.startTransaction();
            table1.select("_id").where("_id = '1'").lockShared().execute();

            // session2.startTransaction();
            // res = table2.select("_id").where("_id < '3'").lockExclusive().execute(); (Can't test)
            // session2.rollback();

            session2.startTransaction();
            futRes = table2.select("_id").where("_id < '3'").lockExclusive().executeAsync();
            final CompletableFuture<RowResult> fr1 = futRes;
            assertThrows(TimeoutException.class, () -> fr1.get(3, TimeUnit.SECONDS));

            session1.rollback(); // Unlocks session2.

            res = futRes.get(3, TimeUnit.SECONDS);
            assertTrue(futRes.isDone());
            assertEquals(2, asStringList.apply(res).size());
            assertThat(asStringList.apply(res), hasItems("1", "2"));
            session2.rollback();

            // session2.lockExclusive(NOWAIT) should return locking error.
            session1.startTransaction();
            table1.select("_id").where("_id = '1'").lockShared().execute();

            session2.startTransaction();
            assertThrows(XProtocolError.class,
                    "ERROR 3572 \\(HY000\\) Statement aborted because lock\\(s\\) could not be acquired immediately and NOWAIT is set\\.",
                    () -> table2.select("_id").where("_id < '3'").lockExclusive(Statement.LockContention.NOWAIT).execute());
            session2.rollback();

            session2.startTransaction();
            futRes = table2.select("_id").where("_id < '3'").lockExclusive(Statement.LockContention.NOWAIT).executeAsync();
            final CompletableFuture<RowResult> fr2 = futRes;
            assertThrows(ExecutionException.class,
                    ".*XProtocolError: ERROR 3572 \\(HY000\\) Statement aborted because lock\\(s\\) could not be acquired immediately and NOWAIT is set\\.",
                    () -> fr2.get(3, TimeUnit.SECONDS));
            session2.rollback();

            session1.rollback();

            // session2.lockExclusive(SKIP_LOCK) should return (unlocked) data immediately.
            session1.startTransaction();
            table1.select("_id").where("_id = '1'").lockShared().execute();

            session2.startTransaction();
            res = table2.select("_id").where("_id < '3'").lockExclusive(Statement.LockContention.SKIP_LOCKED).execute();
            assertEquals(1, asStringList.apply(res).size());
            assertThat(asStringList.apply(res), hasItems("2"));
            session2.rollback();

            session2.startTransaction();
            futRes = table2.select("_id").where("_id < '3'").lockExclusive(Statement.LockContention.SKIP_LOCKED).executeAsync();
            res = futRes.get(3, TimeUnit.SECONDS);
            assertTrue(futRes.isDone());
            assertEquals(1, asStringList.apply(res).size());
            assertThat(asStringList.apply(res), hasItems("2"));
            session2.rollback();

            session1.rollback();

            /*
             * 3. Exclusive Lock in first session and shared lock in second.
             */

            // session2.lockShared() blocks until session1 ends.
            session1.startTransaction();
            table1.select("_id").where("_id = '1'").lockExclusive().execute();

            // session2.startTransaction();
            // res = table2.select("_id").where("_id < '3'").lockShared().execute(); (Can't test)
            // session2.rollback();

            session2.startTransaction();
            futRes = table2.select("_id").where("_id < '3'").lockShared().executeAsync();
            final CompletableFuture<RowResult> fr3 = futRes;
            assertThrows(TimeoutException.class, () -> fr3.get(3, TimeUnit.SECONDS));

            session1.rollback(); // Unlocks session2.

            res = futRes.get(3, TimeUnit.SECONDS);
            assertTrue(futRes.isDone());
            assertEquals(2, asStringList.apply(res).size());
            assertThat(asStringList.apply(res), hasItems("1", "2"));
            session2.rollback();

            // session2.lockShared(NOWAIT) should return locking error.
            session1.startTransaction();
            table1.select("_id").where("_id = '1'").lockExclusive().execute();

            session2.startTransaction();
            assertThrows(XProtocolError.class,
                    "ERROR 3572 \\(HY000\\) Statement aborted because lock\\(s\\) could not be acquired immediately and NOWAIT is set\\.",
                    () -> table2.select("_id").where("_id < '3'").lockShared(Statement.LockContention.NOWAIT).execute());
            session2.rollback();

            session2.startTransaction();
            futRes = table2.select("_id").where("_id < '3'").lockShared(Statement.LockContention.NOWAIT).executeAsync();
            final CompletableFuture<RowResult> fr4 = futRes;
            assertThrows(ExecutionException.class,
                    ".*XProtocolError: ERROR 3572 \\(HY000\\) Statement aborted because lock\\(s\\) could not be acquired immediately and NOWAIT is set\\.",
                    () -> fr4.get(3, TimeUnit.SECONDS));
            session2.rollback();

            session1.rollback();

            // session2.lockShared(SKIP_LOCK) should return (unlocked) data immediately.
            session1.startTransaction();
            table1.select("_id").where("_id = '1'").lockExclusive().execute();

            session2.startTransaction();
            res = table2.select("_id").where("_id < '3'").lockShared(Statement.LockContention.SKIP_LOCKED).execute();
            assertEquals(1, asStringList.apply(res).size());
            assertThat(asStringList.apply(res), hasItems("2"));
            session2.rollback();

            session2.startTransaction();
            futRes = table2.select("_id").where("_id < '3'").lockShared(Statement.LockContention.SKIP_LOCKED).executeAsync();
            res = futRes.get(3, TimeUnit.SECONDS);
            assertTrue(futRes.isDone());
            assertEquals(1, asStringList.apply(res).size());
            assertThat(asStringList.apply(res), hasItems("2"));
            session2.rollback();

            session1.rollback();

            /*
             * 4. Exclusive Lock in both sessions.
             */

            // session2.lockExclusive() blocks until session1 ends.
            session1.startTransaction();
            table1.select("_id").where("_id = '1'").lockExclusive().execute();

            // session2.startTransaction();
            // res = table2.select("_id").where("_id < '3'").lockExclusive().execute(); (Can't test)
            // session2.rollback();

            session2.startTransaction();
            futRes = table2.select("_id").where("_id < '3'").lockExclusive().executeAsync();
            final CompletableFuture<RowResult> fr5 = futRes;
            assertThrows(TimeoutException.class, () -> fr5.get(3, TimeUnit.SECONDS));

            session1.rollback(); // Unlocks session2.

            res = futRes.get(3, TimeUnit.SECONDS);
            assertTrue(futRes.isDone());
            assertEquals(2, asStringList.apply(res).size());
            assertThat(asStringList.apply(res), hasItems("1", "2"));
            session2.rollback();

            // session2.lockExclusive(NOWAIT) should return locking error.
            session1.startTransaction();
            table1.select("_id").where("_id = '1'").lockExclusive().execute();

            session2.startTransaction();
            assertThrows(XProtocolError.class,
                    "ERROR 3572 \\(HY000\\) Statement aborted because lock\\(s\\) could not be acquired immediately and NOWAIT is set\\.",
                    () -> table2.select("_id").where("_id < '3'").lockExclusive(Statement.LockContention.NOWAIT).execute());
            session2.rollback();

            session2.startTransaction();
            futRes = table2.select("_id").where("_id < '3'").lockExclusive(Statement.LockContention.NOWAIT).executeAsync();
            final CompletableFuture<RowResult> fr6 = futRes;
            assertThrows(ExecutionException.class,
                    ".*XProtocolError: ERROR 3572 \\(HY000\\) Statement aborted because lock\\(s\\) could not be acquired immediately and NOWAIT is set\\.",
                    () -> fr6.get(3, TimeUnit.SECONDS));
            session2.rollback();

            session1.rollback();

            // session2.lockExclusive(SKIP_LOCK) should return (unlocked) data immediately.
            session1.startTransaction();
            table1.select("_id").where("_id = '1'").lockExclusive().execute();

            session2.startTransaction();
            res = table2.select("_id").where("_id < '3'").lockExclusive(Statement.LockContention.SKIP_LOCKED).execute();
            assertEquals(1, asStringList.apply(res).size());
            assertThat(asStringList.apply(res), hasItems("2"));
            session2.rollback();

            session2.startTransaction();
            futRes = table2.select("_id").where("_id < '3'").lockExclusive(Statement.LockContention.SKIP_LOCKED).executeAsync();
            res = futRes.get(3, TimeUnit.SECONDS);
            assertTrue(futRes.isDone());
            assertEquals(1, asStringList.apply(res).size());
            assertThat(asStringList.apply(res), hasItems("2"));
            session2.rollback();

            session1.rollback();
        } finally {
            if (session1 != null) {
                session1.close();
            }
            if (session2 != null) {
                session2.close();
            }
            sqlUpdate("DROP TABLE IF EXISTS testTableRowLockOptions");
        }
    }

    /**
     * Tests fix for Bug#22038729, X DEVAPI: ANY API CALL AFTER A FAILED CALL PROC() RESULTS IN HANG
     * and for duplicate Bug#25575010, X DEVAPI: ANY API CALL AFTER A FAILED SELECT RESULTS IN HANG
     *
     * @throws Exception
     */
    @Test
    public void testBug22038729() throws Exception {
        final Field pf = CoreSession.class.getDeclaredField("protocol");
        pf.setAccessible(true);

        try {
            sqlUpdate("drop table if exists testBug22038729");
            sqlUpdate("create table testBug22038729 (c1 int, c2 int unsigned, id bigint)");
            sqlUpdate("insert into testBug22038729 values(10, 100, -9223372036854775808)");
            sqlUpdate("insert into testBug22038729 values(11, 11, 9223372036854775806)");

            sqlUpdate("drop procedure if exists testBug22038729p");
            sqlUpdate("create procedure testBug22038729p (in p1 int,IN p2 char(20)) begin select -10;select id+1000 from testBug22038729; end;");

            // XProtocol.readRowOrNull()
            Session sess = new SessionFactory().getSession(this.testProperties);
            Table t1 = sess.getDefaultSchema().getTable("testBug22038729");
            RowResult rows = t1.select("c1-c2").orderBy("c1 DESC").execute();
            assertTrue(rows.hasNext());
            Row r = rows.next();
            assertEquals(0, r.getInt(0));
            assertThrows(XProtocolError.class, "ERROR 1690 \\(22003\\) BIGINT UNSIGNED value is out of range .*", () -> rows.hasNext());
            sess.close(); // It was hanging

            // XProtocol.readRowOrNull()
            sess = new SessionFactory().getSession(this.testProperties);
            SqlResult rs1 = sess.sql("select c1-c2 from testBug22038729 order by c1 desc").execute();
            assertEquals(0, rs1.fetchOne().getInt(0));
            assertThrows(XProtocolError.class, "ERROR 1690 \\(22003\\) BIGINT UNSIGNED value is out of range .*", () -> rs1.fetchOne());
            sess.close(); // It was hanging

            // XProtocol.drainRows()
            sess = new SessionFactory().getSession(this.testProperties);
            sess.sql("select c1-c2 from testBug22038729 order by c1 desc").execute();
            XProtocol xp = (XProtocol) pf.get(((SessionImpl) sess).getSession());
            assertThrows(XProtocolError.class, "ERROR 1690 \\(22003\\) BIGINT UNSIGNED value is out of range .*", () -> {
                xp.drainRows();
                return xp;
            });
            sess.close(); // It was hanging

            sess = new SessionFactory().getSession(this.testProperties);
            SqlResult rs2 = sess.sql("call testBug22038729p(?, ?)").bind(10).bind("X").execute();
            assertTrue(rs2.hasData());
            assertTrue(rs2.hasNext());
            r = rs2.next();
            assertEquals(-10, r.getInt(0));
            assertFalse(rs2.hasNext());
            assertTrue(rs2.nextResult());
            assertTrue(rs2.hasData());
            assertTrue(rs2.hasNext());
            r = rs2.next();
            assertEquals(-9223372036854774808L, r.getLong(0));
            assertThrows(XProtocolError.class, "ERROR 1690 \\(22003\\) BIGINT value is out of range .*", () -> rs2.hasNext());
            sess.close(); // It was hanging

        } finally {
            sqlUpdate("drop table if exists testBug22038729");
            sqlUpdate("drop procedure if exists testBug22038729p");
        }
    }

    @Test
    public void testPreparedStatements() {
        assumeTrue(mysqlVersionMeetsMinimum(ServerVersion.parseVersion("8.0.14")), "MySQL 8.0.14+ is required to run this test.");

        try {
            // Prepare test data.
            sqlUpdate("DROP TABLE IF EXISTS testPrepareSelect");
            sqlUpdate("CREATE TABLE testPrepareSelect (id INT PRIMARY KEY, ord INT)");
            sqlUpdate("INSERT INTO testPrepareSelect VALUES (1, 1), (2, 2), (3, 3), (4, 4), (5, 5), (6, 6), (7, 7), (8, 8)");

            SessionFactory sf = new SessionFactory();
            /*
             * Test common usage.
             */
            Session testSession = sf.getSession(this.testProperties);

            int sessionThreadId = getThreadId(testSession);
            assertPreparedStatementsCount(sessionThreadId, 0, 1);
            assertPreparedStatementsStatusCounts(testSession, 0, 0, 0);

            Table testTbl = testSession.getDefaultSchema().getTable("testPrepareSelect");

            // Initialize several SelectStatement objects.
            SelectStatement testSelect1 = testTbl.select("ord"); // Select all.
            SelectStatement testSelect2 = testTbl.select("ord").where("ord >= :n"); // Criteria with one placeholder.
            SelectStatement testSelect3 = testTbl.select("ord").where("ord >= :n AND ord <= :n + 3"); // Criteria with same placeholder repeated.
            SelectStatement testSelect4 = testTbl.select("ord").where("ord >= :n AND ord <= :m"); // Criteria with multiple placeholders.

            assertPreparedStatementsCountsAndId(testSession, 0, testSelect1, 0, -1);
            assertPreparedStatementsCountsAndId(testSession, 0, testSelect2, 0, -1);
            assertPreparedStatementsCountsAndId(testSession, 0, testSelect3, 0, -1);
            assertPreparedStatementsCountsAndId(testSession, 0, testSelect4, 0, -1);

            assertPreparedStatementsStatusCounts(testSession, 0, 0, 0);

            // A. Set binds: 1st execute -> non-prepared.
            assertTestPreparedStatementsResult(testSelect1.execute(), 1, 8);
            assertPreparedStatementsCountsAndId(testSession, 0, testSelect1, 0, -1);
            assertTestPreparedStatementsResult(testSelect2.bind("n", 2).execute(), 2, 8);
            assertPreparedStatementsCountsAndId(testSession, 0, testSelect2, 0, -1);
            assertTestPreparedStatementsResult(testSelect3.bind("n", 2).execute(), 2, 5);
            assertPreparedStatementsCountsAndId(testSession, 0, testSelect3, 0, -1);
            assertTestPreparedStatementsResult(testSelect4.bind("n", 2).bind("m", 5).execute(), 2, 5);
            assertPreparedStatementsCountsAndId(testSession, 0, testSelect4, 0, -1);

            assertPreparedStatementsStatusCounts(testSession, 0, 0, 0);

            // B. Set orderBy resets execution count: 1st execute -> non-prepared.
            assertTestPreparedStatementsResult(testSelect1.orderBy("id").execute(), 1, 8);
            assertPreparedStatementsCountsAndId(testSession, 0, testSelect1, 0, -1);
            assertTestPreparedStatementsResult(testSelect2.orderBy("id").execute(), 2, 8);
            assertPreparedStatementsCountsAndId(testSession, 0, testSelect2, 0, -1);
            assertTestPreparedStatementsResult(testSelect3.orderBy("id").execute(), 2, 5);
            assertPreparedStatementsCountsAndId(testSession, 0, testSelect3, 0, -1);
            assertTestPreparedStatementsResult(testSelect4.orderBy("id").execute(), 2, 5);
            assertPreparedStatementsCountsAndId(testSession, 0, testSelect4, 0, -1);

            assertPreparedStatementsStatusCounts(testSession, 0, 0, 0);

            // C. Set binds reuse statement: 2nd execute -> prepare + execute.
            assertTestPreparedStatementsResult(testSelect1.execute(), 1, 8);
            assertPreparedStatementsCountsAndId(testSession, 1, testSelect1, 1, 1);
            assertTestPreparedStatementsResult(testSelect2.bind("n", 3).execute(), 3, 8);
            assertPreparedStatementsCountsAndId(testSession, 2, testSelect2, 2, 1);
            assertTestPreparedStatementsResult(testSelect3.bind("n", 3).execute(), 3, 6);
            assertPreparedStatementsCountsAndId(testSession, 3, testSelect3, 3, 1);
            assertTestPreparedStatementsResult(testSelect4.bind("m", 6).execute(), 2, 6);
            assertPreparedStatementsCountsAndId(testSession, 4, testSelect4, 4, 1);

            assertPreparedStatementsStatusCounts(testSession, 4, 4, 0);

            // D. Set binds reuse statement: 3rd execute -> execute.
            assertTestPreparedStatementsResult(testSelect1.execute(), 1, 8);
            assertPreparedStatementsCountsAndId(testSession, 4, testSelect1, 1, 2);
            assertTestPreparedStatementsResult(testSelect2.bind("n", 4).execute(), 4, 8);
            assertPreparedStatementsCountsAndId(testSession, 4, testSelect2, 2, 2);
            assertTestPreparedStatementsResult(testSelect3.bind("n", 4).execute(), 4, 7);
            assertPreparedStatementsCountsAndId(testSession, 4, testSelect3, 3, 2);
            assertTestPreparedStatementsResult(testSelect4.bind("n", 3).bind("m", 7).execute(), 3, 7);
            assertPreparedStatementsCountsAndId(testSession, 4, testSelect4, 4, 2);

            assertPreparedStatementsStatusCounts(testSession, 4, 8, 0);

            // E. Set where deallocates and resets execution count: 1st execute -> deallocate + non-prepared.
            assertTestPreparedStatementsResult(testSelect1.where("true").execute(), 1, 8);
            assertPreparedStatementsCountsAndId(testSession, 3, testSelect1, 0, -1);
            assertTestPreparedStatementsResult(testSelect2.where("true AND ord >= :n").bind("n", 4).execute(), 4, 8);
            assertPreparedStatementsCountsAndId(testSession, 2, testSelect2, 0, -1);
            assertTestPreparedStatementsResult(testSelect3.where("true AND ord >= :n AND ord <= :n + 3").bind("n", 4).execute(), 4, 7);
            assertPreparedStatementsCountsAndId(testSession, 1, testSelect3, 0, -1);
            assertTestPreparedStatementsResult(testSelect4.where("true AND ord >= :n AND ord <= :m").bind("n", 3).bind("m", 7).execute(), 3, 7);
            assertPreparedStatementsCountsAndId(testSession, 0, testSelect4, 0, -1);

            assertPreparedStatementsStatusCounts(testSession, 4, 8, 4);

            // F. No Changes: 2nd execute -> prepare + execute.
            assertTestPreparedStatementsResult(testSelect1.execute(), 1, 8);
            assertPreparedStatementsCountsAndId(testSession, 1, testSelect1, 1, 1);
            assertTestPreparedStatementsResult(testSelect2.bind("n", 4).execute(), 4, 8);
            assertPreparedStatementsCountsAndId(testSession, 2, testSelect2, 2, 1);
            assertTestPreparedStatementsResult(testSelect3.bind("n", 4).execute(), 4, 7);
            assertPreparedStatementsCountsAndId(testSession, 3, testSelect3, 3, 1);
            assertTestPreparedStatementsResult(testSelect4.bind("n", 3).bind("m", 7).execute(), 3, 7);
            assertPreparedStatementsCountsAndId(testSession, 4, testSelect4, 4, 1);

            assertPreparedStatementsStatusCounts(testSession, 8, 12, 4);

            // G. Set limit for the first time deallocates and re-prepares: 1st execute -> re-prepare + execute.
            assertTestPreparedStatementsResult(testSelect1.limit(2).execute(), 1, 2);
            assertPreparedStatementsCountsAndId(testSession, 4, testSelect1, 1, 1);
            assertTestPreparedStatementsResult(testSelect2.limit(2).execute(), 4, 5);
            assertPreparedStatementsCountsAndId(testSession, 4, testSelect2, 2, 1);
            assertTestPreparedStatementsResult(testSelect3.limit(2).execute(), 4, 5);
            assertPreparedStatementsCountsAndId(testSession, 4, testSelect3, 3, 1);
            assertTestPreparedStatementsResult(testSelect4.limit(2).execute(), 3, 4);
            assertPreparedStatementsCountsAndId(testSession, 4, testSelect4, 4, 1);

            assertPreparedStatementsStatusCounts(testSession, 12, 16, 8);

            // H. Set limit and offset reuse prepared statement: 2nd execute -> execute.
            assertTestPreparedStatementsResult(testSelect1.limit(1).offset(1).execute(), 2, 2);
            assertPreparedStatementsCountsAndId(testSession, 4, testSelect1, 1, 2);
            assertTestPreparedStatementsResult(testSelect2.limit(1).offset(1).execute(), 5, 5);
            assertPreparedStatementsCountsAndId(testSession, 4, testSelect2, 2, 2);
            assertTestPreparedStatementsResult(testSelect3.limit(1).offset(1).execute(), 5, 5);
            assertPreparedStatementsCountsAndId(testSession, 4, testSelect3, 3, 2);
            assertTestPreparedStatementsResult(testSelect4.limit(1).offset(1).execute(), 4, 4);
            assertPreparedStatementsCountsAndId(testSession, 4, testSelect4, 4, 2);

            assertPreparedStatementsStatusCounts(testSession, 12, 20, 8);

            // I. Set orderBy deallocates and resets execution count, set limit and bind has no effect: 1st execute -> deallocate + non-prepared.
            assertTestPreparedStatementsResult(testSelect1.orderBy("id").limit(2).execute(), 2, 3);
            assertPreparedStatementsCountsAndId(testSession, 3, testSelect1, 0, -1);
            assertTestPreparedStatementsResult(testSelect2.orderBy("id").limit(2).bind("n", 4).execute(), 5, 6);
            assertPreparedStatementsCountsAndId(testSession, 2, testSelect2, 0, -1);
            assertTestPreparedStatementsResult(testSelect3.orderBy("id").limit(2).bind("n", 4).execute(), 5, 6);
            assertPreparedStatementsCountsAndId(testSession, 1, testSelect3, 0, -1);
            assertTestPreparedStatementsResult(testSelect4.orderBy("id").limit(2).bind("m", 7).execute(), 4, 5);
            assertPreparedStatementsCountsAndId(testSession, 0, testSelect4, 0, -1);

            assertPreparedStatementsStatusCounts(testSession, 12, 20, 12);

            // J. Set offset reuse statement: 2nd execute -> prepare + execute.
            assertTestPreparedStatementsResult(testSelect1.offset(0).execute(), 1, 2);
            assertPreparedStatementsCountsAndId(testSession, 1, testSelect1, 1, 1);
            assertTestPreparedStatementsResult(testSelect2.offset(0).execute(), 4, 5);
            assertPreparedStatementsCountsAndId(testSession, 2, testSelect2, 2, 1);
            assertTestPreparedStatementsResult(testSelect3.offset(0).execute(), 4, 5);
            assertPreparedStatementsCountsAndId(testSession, 3, testSelect3, 3, 1);
            assertTestPreparedStatementsResult(testSelect4.offset(0).execute(), 3, 4);
            assertPreparedStatementsCountsAndId(testSession, 4, testSelect4, 4, 1);

            assertPreparedStatementsStatusCounts(testSession, 16, 24, 12);

            testSession.close();
            assertPreparedStatementsCount(sessionThreadId, 0, 10); // Prepared statements won't live past the closing of the session.

            /*
             * Test falling back onto non-prepared statements.
             */
            testSession = sf.getSession(this.testProperties);
            int origMaxPrepStmtCount = this.session.sql("SELECT @@max_prepared_stmt_count").execute().fetchOne().getInt(0);

            try {
                // Allow preparing only one more statement.
                this.session.sql("SET GLOBAL max_prepared_stmt_count = ?").bind(getPreparedStatementsCount() + 1).execute();

                sessionThreadId = getThreadId(testSession);
                assertPreparedStatementsCount(sessionThreadId, 0, 1);
                assertPreparedStatementsStatusCounts(testSession, 0, 0, 0);

                testTbl = testSession.getDefaultSchema().getTable("testPrepareSelect");

                testSelect1 = testTbl.select("ord");
                testSelect2 = testTbl.select("ord");

                // 1st execute -> don't prepare.
                assertTestPreparedStatementsResult(testSelect1.execute(), 1, 8);
                assertPreparedStatementsCountsAndId(testSession, 0, testSelect1, 0, -1);
                assertTestPreparedStatementsResult(testSelect2.execute(), 1, 8);
                assertPreparedStatementsCountsAndId(testSession, 0, testSelect2, 0, -1);

                assertPreparedStatementsStatusCounts(testSession, 0, 0, 0);

                // 2nd execute -> prepare + execute.
                assertTestPreparedStatementsResult(testSelect1.execute(), 1, 8);
                assertPreparedStatementsCountsAndId(testSession, 1, testSelect1, 1, 1);
                assertTestPreparedStatementsResult(testSelect2.execute(), 1, 8); // Fails preparing, execute as non-prepared.
                assertPreparedStatementsCountsAndId(testSession, 1, testSelect2, 0, -1);

                assertPreparedStatementsStatusCounts(testSession, 2, 1, 0); // Failed prepare also counts.

                // 3rd execute -> execute.
                assertTestPreparedStatementsResult(testSelect1.execute(), 1, 8);
                assertPreparedStatementsCountsAndId(testSession, 1, testSelect1, 1, 2);
                assertTestPreparedStatementsResult(testSelect2.execute(), 1, 8); // Execute as non-prepared.
                assertPreparedStatementsCountsAndId(testSession, 1, testSelect2, 0, -1);

                assertPreparedStatementsStatusCounts(testSession, 2, 2, 0);

                testSession.close();
                assertPreparedStatementsCount(sessionThreadId, 0, 10); // Prepared statements won't live past the closing of the session.
            } finally {
                this.session.sql("SET GLOBAL max_prepared_stmt_count = ?").bind(origMaxPrepStmtCount).execute();
            }
        } finally {
            sqlUpdate("DROP TABLE IF EXISTS testPrepareSelect");
        }
    }

    private void assertTestPreparedStatementsResult(RowResult res, int expectedMin, int expectedMax) {
        for (Row r : res.fetchAll()) {
            assertEquals(expectedMin++, r.getInt("ord"));
        }
        assertEquals(expectedMax, expectedMin - 1);
    }

    /**
     * Tests server Bug#31667405, INCORRECT PREPARED STATEMENT OUTCOME WITH NUMERIC STRINGS IN JSON.
     */
    @Test
    public void testBug31667405() {
        assumeTrue(mysqlVersionMeetsMinimum(ServerVersion.parseVersion("8.0.14")), "MySQL 8.0.14+ is required to run this test.");

        try {
            // Prepare test data.
            sqlUpdate("DROP TABLE IF EXISTS testBug31667405");
            sqlUpdate("CREATE TABLE testBug31667405 (doc JSON)");
            sqlUpdate("INSERT INTO testBug31667405 VALUES ('{\"ord\":\"1\"}')");

            SessionFactory sf = new SessionFactory();
            Session testSession = sf.getSession(this.testProperties);

            int sessionThreadId = getThreadId(testSession);
            assertPreparedStatementsCount(sessionThreadId, 0, 1);
            assertPreparedStatementsStatusCounts(testSession, 0, 0, 0);

            Table testTbl = testSession.getDefaultSchema().getTable("testBug31667405");

            // Initialize SelectStatement object.
            SelectStatement testSelect2 = testTbl.select().where("$.ord == :n");
            assertPreparedStatementsCountsAndId(testSession, 0, testSelect2, 0, -1);
            assertPreparedStatementsStatusCounts(testSession, 0, 0, 0);

            // A. Set binds: 1st execute -> non-prepared.
            testSelect2.bind("n", "1").execute();
            assertPreparedStatementsCountsAndId(testSession, 0, testSelect2, 0, -1);
            assertPreparedStatementsStatusCounts(testSession, 0, 0, 0);

            // B. Set binds reuse statement: 2nd execute -> prepare + execute.
            RowResult res = testSelect2.bind("n", "1").execute();
            assertEquals(1, res.count());

            assertPreparedStatementsCountsAndId(testSession, 1, testSelect2, 1, 1);
            assertPreparedStatementsStatusCounts(testSession, 1, 1, 0);
        } finally {
            sqlUpdate("DROP TABLE IF EXISTS testBug31667405");
        }
    }

}
