Fix issue #50

When a statement contains several queries separated by a semicolon character
and at least one of these queries contains a string value including a semicolon
the list of queries to execute was incorrect resulting into a syntax error
during the execution of the split statements.

The statements splitting has been refactored and is applied in any compliance
mode.

Author: maximevw

Diff for: CHANGELOG.md

@@ -4,6 +4,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) and this project adheres to 
 [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [4.11.1] - 2023-12-30
+### Fixed
+- Fix issue [#50](https://github.com/ing-bank/cassandra-jdbc-wrapper/issues/50) preventing a correct execution of
+  multiple statements separated by semicolon characters (`;`) when at least one of the CQL queries contains a semicolon
+  character which is not a query separator.
+
 ## [4.11.0] - 2023-12-03
 ### Added
 - Add support for connections with multiple contact points using different ports (see feature request 
@@ -219,6 +225,7 @@ For this version, the changelog lists the main changes comparatively to the late
 - Fix logs in `CassandraConnection` constructor.
 
 [original project]: https://github.com/adejanovski/cassandra-jdbc-wrapper/
+[4.11.1]: https://github.com/ing-bank/cassandra-jdbc-wrapper/compare/v4.11.0...v4.11.1
 [4.11.0]: https://github.com/ing-bank/cassandra-jdbc-wrapper/compare/v4.10.2...v4.11.0
 [4.10.2]: https://github.com/ing-bank/cassandra-jdbc-wrapper/compare/v4.10.1...v4.10.2
 [4.10.1]: https://github.com/ing-bank/cassandra-jdbc-wrapper/compare/v4.10.0...v4.10.1

Diff for: pom.xml

@@ -5,7 +5,7 @@
 
 	<groupId>com.ing.data</groupId>
 	<artifactId>cassandra-jdbc-wrapper</artifactId>
-    <version>4.11.0</version>
+    <version>4.11.1</version>
 	<packaging>jar</packaging>
 
     <name>Cassandra JDBC Wrapper</name>

Diff for: src/main/java/com/ing/data/cassandra/jdbc/CassandraPreparedStatement.java

@@ -79,6 +79,7 @@
 import static com.ing.data.cassandra.jdbc.utils.ConversionsUtil.convertToLocalDate;
 import static com.ing.data.cassandra.jdbc.utils.ConversionsUtil.convertToLocalTime;
 import static com.ing.data.cassandra.jdbc.utils.ErrorConstants.NO_RESULT_SET;
+import static com.ing.data.cassandra.jdbc.utils.ErrorConstants.TOO_MANY_QUERIES;
 import static com.ing.data.cassandra.jdbc.utils.ErrorConstants.UNSUPPORTED_JDBC_TYPE;
 import static com.ing.data.cassandra.jdbc.utils.ErrorConstants.VECTOR_ELEMENTS_NOT_NUMBERS;
 import static com.ing.data.cassandra.jdbc.utils.JsonUtil.getObjectMapper;
@@ -233,8 +234,7 @@ public void addBatch() throws SQLException {
         this.batchStatements.add(this.boundStatement);
         this.boundStatement = this.preparedStatement.boundStatementBuilder().build();
         if (this.batchStatements.size() > MAX_ASYNC_QUERIES) {
-            throw new SQLNonTransientException("Too many queries at once (" + batchStatements.size() + "). You must "
-                + "split your queries into more batches!");
+            throw new SQLNonTransientException(String.format(TOO_MANY_QUERIES, batchStatements.size()));
         }
     }
 

Diff for: src/main/java/com/ing/data/cassandra/jdbc/CassandraStatement.java

@@ -24,7 +24,6 @@
 import com.datastax.oss.driver.internal.core.cql.MultiPageResultSet;
 import com.datastax.oss.driver.internal.core.cql.SinglePageResultSet;
 import com.datastax.oss.driver.internal.core.util.concurrent.CompletableFutures;
-import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -58,7 +57,9 @@
 import static com.ing.data.cassandra.jdbc.utils.ErrorConstants.NO_GEN_KEYS;
 import static com.ing.data.cassandra.jdbc.utils.ErrorConstants.NO_MULTIPLE;
 import static com.ing.data.cassandra.jdbc.utils.ErrorConstants.NO_RESULT_SET;
+import static com.ing.data.cassandra.jdbc.utils.ErrorConstants.TOO_MANY_QUERIES;
 import static com.ing.data.cassandra.jdbc.utils.ErrorConstants.WAS_CLOSED_STMT;
+import static org.apache.commons.lang3.StringUtils.countMatches;
 
 /**
  * Cassandra statement: implementation class for {@link Statement}.
@@ -292,21 +293,50 @@ public int compareTo(@Nonnull final Object target) {
         return 1;
     }
 
+    private List<String> splitStatements(final String cql) {
+        final String[] cqlQueries = cql.split(STATEMENTS_SEPARATOR_REGEX);
+        final List<String> cqlQueriesToExecute = new ArrayList<>(cqlQueries.length);
+
+        // If a query contains a semicolon character in a string value (for example 'abc;xyz'), re-merge queries
+        // wrongly split.
+        final String singleQuote = "'";
+        StringBuilder prevCqlQuery = new StringBuilder();
+        for (final String cqlQuery : cqlQueries) {
+            final boolean hasStringValues = cqlQuery.contains(singleQuote);
+            final boolean isFirstQueryPartWithIncompleteStringValue =
+                countMatches(cqlQuery, singleQuote) % 2 == 1 && prevCqlQuery.length() == 0;
+            final boolean isNotFirstQueryPartWithCompleteStringValue =
+                countMatches(cqlQuery, singleQuote) % 2 == 0 && prevCqlQuery.length() > 0;
+            final boolean isNotFirstQueryPartWithoutStringValue =
+                !prevCqlQuery.toString().isEmpty() && !cqlQuery.contains(singleQuote);
+
+            if ((hasStringValues && (isFirstQueryPartWithIncompleteStringValue
+                || isNotFirstQueryPartWithCompleteStringValue)) || isNotFirstQueryPartWithoutStringValue) {
+                prevCqlQuery.append(cqlQuery).append(";");
+            } else {
+                prevCqlQuery.append(cqlQuery);
+                cqlQueriesToExecute.add(prevCqlQuery.toString());
+                prevCqlQuery = new StringBuilder();
+            }
+        }
+        return cqlQueriesToExecute;
+    }
+
     private void doExecute(final String cql) throws SQLException {
         final List<CompletionStage<AsyncResultSet>> futures = new ArrayList<>();
 
         try {
-            final String[] cqlQueries = cql.split(STATEMENTS_SEPARATOR_REGEX);
-            if (cqlQueries.length > 1
+            final List<String> cqlQueries = splitStatements(cql);
+            final int nbQueriesToExecute = cqlQueries.size();
+            if (nbQueriesToExecute > 1
                 && !(cql.trim().toLowerCase().startsWith("begin")
                 && cql.toLowerCase().contains("batch") && cql.toLowerCase().contains("apply"))) {
                 final ArrayList<com.datastax.oss.driver.api.core.cql.ResultSet> results = new ArrayList<>();
 
-                // Several statements in the query to execute asynchronously...
-                if (cqlQueries.length > MAX_ASYNC_QUERIES * 1.1) {
+                // Several statements in the query to execute potentially asynchronously...
+                if (nbQueriesToExecute > MAX_ASYNC_QUERIES * 1.1) {
                     // Protect the cluster from receiving too many queries at once and force the dev to split the load
-                    throw new SQLNonTransientException("Too many queries at once (" + cqlQueries.length
-                        + "). You must split your queries into more batches !");
+                    throw new SQLNonTransientException(String.format(TOO_MANY_QUERIES, nbQueriesToExecute));
                 }
 
                 // If we should not execute the queries asynchronously, for example if they must be executed in the
@@ -318,29 +348,19 @@ private void doExecute(final String cql) throws SQLException {
                         results.add(rs);
                     }
                 } else {
-                    StringBuilder prevCqlQuery = new StringBuilder();
                     for (final String cqlQuery : cqlQueries) {
-                        if ((cqlQuery.contains("'") && ((StringUtils.countMatches(cqlQuery, "'") % 2 == 1
-                            && prevCqlQuery.length() == 0)
-                            || (StringUtils.countMatches(cqlQuery, "'") % 2 == 0 && prevCqlQuery.length() > 0)))
-                            || (!prevCqlQuery.toString().isEmpty() && !cqlQuery.contains("'"))) {
-                            prevCqlQuery.append(cqlQuery).append(";");
-                        } else {
-                            prevCqlQuery.append(cqlQuery);
-                            if (LOG.isTraceEnabled() || this.connection.isDebugMode()) {
-                                LOG.debug("CQL: {}", prevCqlQuery);
-                            }
-                            SimpleStatement stmt = SimpleStatement.newInstance(prevCqlQuery.toString())
-                                .setConsistencyLevel(this.connection.getDefaultConsistencyLevel())
-                                .setPageSize(this.fetchSize);
-                            if (this.customTimeoutProfile != null) {
-                                stmt = stmt.setExecutionProfile(this.customTimeoutProfile);
-                            }
-                            final CompletionStage<AsyncResultSet> resultSetFuture =
-                                ((CqlSession) this.connection.getSession()).executeAsync(stmt);
-                            futures.add(resultSetFuture);
-                            prevCqlQuery = new StringBuilder();
+                        if (LOG.isDebugEnabled() || this.connection.isDebugMode()) {
+                            LOG.debug("CQL: {}", cqlQuery);
+                        }
+                        SimpleStatement stmt = SimpleStatement.newInstance(cqlQuery)
+                            .setConsistencyLevel(this.connection.getDefaultConsistencyLevel())
+                            .setPageSize(this.fetchSize);
+                        if (this.customTimeoutProfile != null) {
+                            stmt = stmt.setExecutionProfile(this.customTimeoutProfile);
                         }
+                        final CompletionStage<AsyncResultSet> resultSetFuture =
+                            ((CqlSession) this.connection.getSession()).executeAsync(stmt);
+                        futures.add(resultSetFuture);
                     }
 
                     for (final CompletionStage<AsyncResultSet> future : futures) {

Diff for: src/main/java/com/ing/data/cassandra/jdbc/utils/ErrorConstants.java

@@ -17,6 +17,7 @@
 
 import com.ing.data.cassandra.jdbc.CassandraPreparedStatement;
 import com.ing.data.cassandra.jdbc.CassandraResultSet;
+import com.ing.data.cassandra.jdbc.CassandraStatement;
 import com.ing.data.cassandra.jdbc.metadata.MetadataRow;
 
 import java.net.URI;
@@ -278,6 +279,16 @@ public final class ErrorConstants {
      */
     public static final String UNABLE_TO_POPULATE_METADATA_ROW = "Unable to populate a metadata row.";
 
+    /**
+     * Error message used in any SQL exception thrown when the number of CQL queries included in a single batch of
+     * statements is greater than the allowed limit ({@value CassandraStatement#MAX_ASYNC_QUERIES} for a prepared batch
+     * statement or 1.1 {@code *} {@value CassandraStatement#MAX_ASYNC_QUERIES} for a split single statement).
+     * This message is a template expecting the number of CQL queries to execute as placeholder (example:
+     * {@code String.format(TOO_MANY_QUERIES, 10000)}).
+     */
+    public static final String TOO_MANY_QUERIES =
+        "Too many queries at once (%d). You must split your queries into more batches!";
+
     private ErrorConstants() {
         // Private constructor to hide the public one.
     }

Diff for: src/test/java/com/ing/data/cassandra/jdbc/BatchStatementsUnitTest.java

@@ -28,6 +28,7 @@
 import java.sql.Statement;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Set;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -247,4 +248,46 @@ void givenBatchSimpleSplitStatementWithErrors_whenExecute_throwException() throw
         }
         assertThrows(SQLTransientException.class, () -> stmt2.execute(queryBuilder.toString()));
     }
+
+    @SuppressWarnings("unchecked")
+    @ParameterizedTest
+    @Order(7)
+    @ValueSource(strings = {"Default", "Liquibase"})
+    void givenStatementWithValuesIncludingSemicolons_whenExecute_returnExpectedResult(final String complianceMode)
+        throws Exception {
+        sqlConnection2 = newConnection(KEYSPACE, "localdatacenter=datacenter1", "compliancemode=" + complianceMode);
+        final Statement truncateStmt = sqlConnection2.createStatement();
+        truncateStmt.execute("TRUNCATE collections_test");
+
+        final Statement statement = sqlConnection2.createStatement();
+        final StringBuilder queryBuilder = new StringBuilder();
+        for (int i = 0; i < 20; i++) {
+            queryBuilder.append("INSERT INTO collections_test (keyValue, setValue) VALUES( ").append(i)
+                .append(", {'test;0', 'val;").append(i).append("'} );");
+        }
+        statement.execute(queryBuilder.toString());
+        statement.close();
+
+        final StringBuilder query = new StringBuilder();
+        for (int i = 0; i < 20; i++) {
+            query.append("SELECT * FROM collections_test WHERE keyValue = ").append(i).append(";");
+        }
+        final Statement selectStatement = sqlConnection2.createStatement();
+        final ResultSet result = selectStatement.executeQuery(query.toString());
+        int nbRowsInResult = 0;
+        final ArrayList<Integer> foundKeyValues = new ArrayList<>();
+        final ArrayList<String> foundSetValues = new ArrayList<>();
+        while (result.next()) {
+            nbRowsInResult++;
+            foundKeyValues.add(result.getInt("keyValue"));
+            final Set<String> setValues = (Set<String>) result.getObject("setValue");
+            foundSetValues.addAll(setValues);
+        }
+        assertEquals(20, nbRowsInResult);
+        for (int i = 0; i < 20; i++) {
+            assertTrue(foundKeyValues.contains(i));
+            assertTrue(foundSetValues.contains("val;" + i));
+        }
+        selectStatement.close();
+    }
 }