Batch statement preparation fails if multiple tables and parameters a…

…re used

The logic for detecting if a batch statement affects multiple tables in BatchStatement#getPartitionKeyBindVariableIndexes is inverted.
BatchTest#testBatchMultipleTablePrepare has been added to demonstrate the issue.

 patch by Bryn Cooke; reviewed by Eduard Tudenhöfner, Robert Stupp for CASSANDRA-15730

CHANGES.txt

@@ -1,4 +1,5 @@
 4.0-alpha4
+ * Fix batch statement preparation when multiple tables and parameters are used (CASSANDRA-15730)
  * Fix regression with traceOutgoingMessage printing message size (CASSANDRA-15687)
  * Ensure repaired data tracking reads a consistent amount of data across replicas (CASSANDRA-15601)
  * Fix CQLSH to avoid arguments being evaluated (CASSANDRA-15660)

src/java/org/apache/cassandra/cql3/statements/BatchStatement.java

@@ -141,7 +141,7 @@ public List<ColumnSpecification> getBindVariables()
     public short[] getPartitionKeyBindVariableIndexes()
     {
         boolean affectsMultipleTables =
-            !statements.isEmpty() && statements.stream().map(s -> s.metadata().id).allMatch(isEqual(statements.get(0).metadata().id));
+            !statements.isEmpty() && !statements.stream().map(s -> s.metadata().id).allMatch(isEqual(statements.get(0).metadata().id));
 
         // Use the TableMetadata of the first statement for partition key bind indexes.  If the statements affect
         // multiple tables, we won't send partition key bind indexes.

test/unit/org/apache/cassandra/cql3/validation/operations/BatchTest.java

@@ -167,6 +167,26 @@ public void testBatchMultipleTable() throws Throwable
         assertRows(execute(String.format("SELECT * FROM %s", tbl2)), row(0, 3, 4));
     }
 
+    @Test
+    public void testBatchMultipleTablePrepare() throws Throwable
+    {
+        String tbl1 = KEYSPACE + "." + createTableName();
+        String tbl2 = KEYSPACE + "." + createTableName();
+
+        schemaChange(String.format("CREATE TABLE %s (k1 int PRIMARY KEY, v1 int)", tbl1));
+        schemaChange(String.format("CREATE TABLE %s (k2 int PRIMARY KEY, v2 int)", tbl2));
+
+        String query = "BEGIN BATCH " +
+                   String.format("UPDATE %s SET v1 = 1 WHERE k1 = ?;", tbl1) +
+                   String.format("UPDATE %s SET v2 = 2 WHERE k2 = ?;", tbl2) +
+                   "APPLY BATCH;";
+        prepare(query);
+        execute(query, 0, 1);
+
+        assertRows(execute(String.format("SELECT * FROM %s", tbl1)), row(0, 1));
+        assertRows(execute(String.format("SELECT * FROM %s", tbl2)), row(1, 2));
+    }
+
     @Test
     public void testBatchWithInRestriction() throws Throwable
     {