Merge pull request #2549 from potoo0/issue-2548

[JDBC-V2] fix(`StatementImpl`/`PreparedStatementImpl`): prevent duplicate data insertion by clearing batch after execution for issue-2548

Author: chernser

examples/jdbc/src/main/java/com/clickhouse/examples/jdbc/Basic.java

@@ -59,13 +59,20 @@ static void insertDateWithPreparedStatement(String url, Properties properties) t
                 pstmt.setString(3, "Alice");//Set the third parameter to "Alice"
                 pstmt.setObject(4, Collections.singletonMap("key1", "value1"));
                 pstmt.addBatch();//Add the current parameters to the batch
+                pstmt.executeBatch();//Execute the batch
 
                 pstmt.setObject(1, ZonedDateTime.now());
                 pstmt.setInt(2, 2);//Set the second parameter to 2
                 pstmt.setString(3, "Bob");//Set the third parameter to "Bob"
                 pstmt.setObject(4, Collections.singletonMap("key2", "value2"));
                 pstmt.addBatch();//Add the current parameters to the batch
 
+                pstmt.setObject(1, ZonedDateTime.now());
+                pstmt.setInt(2, 2);//Set the second parameter to 2
+                pstmt.setString(3, "Chris");//Set the third parameter to "Chris"
+                pstmt.setObject(4, Collections.singletonMap("key3", "value3"));
+                pstmt.addBatch();//Add the current parameters to the batch
+
                 pstmt.executeBatch();//Execute the batch
             }
         }

jdbc-v2/src/main/java/com/clickhouse/jdbc/PreparedStatementImpl.java

@@ -332,15 +332,23 @@ public long[] executeLargeBatch() throws SQLException {
     }
 
     private List<Integer> executeBatchImpl() throws SQLException {
+        List<Integer> results;
         if (insertStmtWithValues) {
-            return executeInsertBatch();
+            results = executeInsertBatch();
         } else {
-            List<Integer> results = new ArrayList<>();
+            results = new ArrayList<>();
             for (String sql : batch) {
                 results.add((int) executeUpdateImpl(sql, localSettings));
             }
-            return results;
         }
+        clearBatch();
+        return results;
+    }
+
+    @Override
+    public void clearBatch() throws SQLException {
+        super.clearBatch(); /// clear super#batch
+        batchValues.clear();
     }
 
     private List<Integer> executeInsertBatch() throws SQLException {

jdbc-v2/src/main/java/com/clickhouse/jdbc/StatementImpl.java

@@ -416,6 +416,7 @@ private List<Integer> executeBatchImpl() throws SQLException {
         for (String sql : batch) {
             results.add(executeUpdate(sql));
         }
+        clearBatch();
         return results;
     }
 

jdbc-v2/src/test/java/com/clickhouse/jdbc/PreparedStatementTest.java

@@ -953,6 +953,114 @@ void testBatchInsertTextStatement(String sql) throws Exception {
         }
     }
 
+    @Test(groups = {"integration"})
+    void testBatchInsertNoValuesReuse() throws Exception {
+        String table = "test_pstmt_batch_novalues_reuse";
+        String sql = "INSERT INTO %s (v1, v2) VALUES (?, ?)";
+        long seed = System.currentTimeMillis();
+        Random rnd = new Random(seed);
+        try (Connection conn = getJdbcConnection()) {
+
+            try (Statement stmt = conn.createStatement()) {
+                stmt.execute("CREATE TABLE IF NOT EXISTS " + table +
+                        " (v1 Int32, v2 Int32) Engine MergeTree ORDER BY ()");
+            }
+
+            final int nBatches = 10;
+            try (PreparedStatement stmt = conn.prepareStatement(String.format(sql, table))) {
+                Assert.assertEquals(stmt.getClass(), PreparedStatementImpl.class);
+                // add a batch with invalid values
+                stmt.setString(1, "invalid");
+                stmt.setInt(2, rnd.nextInt());
+                stmt.addBatch();
+                assertThrows(SQLException.class, stmt::executeBatch);
+                // should fail due to the previous batch data.
+                assertThrows(SQLException.class, stmt::executeBatch);
+                // clear previous batch data
+                stmt.clearBatch();
+
+                for (int step = 0; step < 2; step++) {
+                    for (int bI = 0; bI < (nBatches >> 1); bI++) {
+                        stmt.setInt(1, rnd.nextInt());
+                        stmt.setInt(2, rnd.nextInt());
+                        stmt.addBatch();
+                    }
+
+                    // reuse the same statement
+                    int[] result = stmt.executeBatch();
+                    for (int r : result) {
+                        Assert.assertEquals(r, 1);
+                    }
+                }
+            }
+
+            try (Statement stmt = conn.createStatement();
+                 ResultSet rs = stmt.executeQuery("SELECT * FROM " + table);) {
+
+                int count = 0;
+                while (rs.next()) {
+                    assertTrue(rs.getInt(1) != 0);
+                    assertTrue(rs.getInt(2) != 0);
+                    count++;
+                }
+                assertEquals(count, nBatches);
+            }
+        }
+    }
+
+    @Test()
+    void testBatchInsertValuesReuse() throws Exception {
+        String table = "test_pstmt_batch_values_reuse";
+        String sql = "INSERT INTO %s (v1, v2) VALUES (1, ?)";
+        long seed = System.currentTimeMillis();
+        Random rnd = new Random(seed);
+        try (Connection conn = getJdbcConnection()) {
+
+            try (Statement stmt = conn.createStatement()) {
+                stmt.execute("CREATE TABLE IF NOT EXISTS " + table +
+                        " (v1 Int32, v2 Int32) Engine MergeTree ORDER BY ()");
+            }
+
+            final int nBatches = 10;
+            try (PreparedStatement stmt = conn.prepareStatement(String.format(sql, table))) {
+                Assert.assertEquals(stmt.getClass(), PreparedStatementImpl.class);
+                // add a batch with invalid values
+                stmt.setString(1, "invalid");
+                stmt.addBatch();
+                assertThrows(SQLException.class, stmt::executeBatch);
+                // should fail due to the previous batch data.
+                assertThrows(SQLException.class, stmt::executeBatch);
+                // clear previous batch data
+                stmt.clearBatch();
+
+                for (int step = 0; step < 2; step++) {
+                    for (int bI = 0; bI < (nBatches >> 1); bI++) {
+                        stmt.setInt(1, rnd.nextInt());
+                        stmt.addBatch();
+                    }
+
+                    // reuse the same statement
+                    int[] result = stmt.executeBatch();
+                    for (int r : result) {
+                        Assert.assertEquals(r, 1);
+                    }
+                }
+            }
+
+            try (Statement stmt = conn.createStatement();
+                 ResultSet rs = stmt.executeQuery("SELECT * FROM " + table);) {
+
+                int count = 0;
+                while (rs.next()) {
+                    assertTrue(rs.getInt(1) != 0);
+                    assertTrue(rs.getInt(2) != 0);
+                    count++;
+                }
+                assertEquals(count, nBatches);
+            }
+        }
+    }
+
     @Test(groups = {"integration"})
     void testWriteUUID() throws Exception {
         String sql = "insert into `test_issue_2327` (`id`, `uuid`) values (?, ?)";

jdbc-v2/src/test/java/com/clickhouse/jdbc/StatementTest.java

@@ -290,6 +290,44 @@ public void testExecuteUpdateBatch() throws Exception {
         }
     }
 
+    @Test(groups = {"integration"})
+    public void testExecuteUpdateBatchReuse() throws Exception {
+        String tableClause = getDatabase() + ".batch_reuse";
+        try (Connection conn = getJdbcConnection()) {
+            try (Statement stmt = conn.createStatement()) {
+                assertEquals(stmt.executeUpdate("CREATE TABLE IF NOT EXISTS " + tableClause + " (id UInt8, num UInt8) ENGINE = MergeTree ORDER BY ()"), 0);
+                // add and execute first invalid batch
+                stmt.addBatch("INSERT INTO " + tableClause + " VALUES (0, 'invalid')");
+                assertThrows(SQLException.class, stmt::executeBatch);
+
+                // add and execute second batch, which should fail due to the previous batch data.
+                stmt.addBatch("INSERT INTO " + tableClause + " VALUES (1, 2)");
+                assertThrows(SQLException.class, stmt::executeBatch);
+
+                // add and execute third batch, which should not fail
+                stmt.clearBatch();
+                stmt.addBatch("INSERT INTO " + tableClause + " VALUES (0, 1)");
+                stmt.addBatch("INSERT INTO " + tableClause + " VALUES (1, 2)");
+                assertEquals(stmt.executeBatch(), new int[]{1, 1});
+
+                stmt.addBatch("INSERT INTO " + tableClause + " VALUES (2, 3), (3, 4)");
+                assertEquals(stmt.executeBatch(), new int[]{2});
+
+                try (ResultSet rs = stmt.executeQuery("SELECT num FROM " + tableClause + " ORDER BY id")) {
+                    assertTrue(rs.next());
+                    assertEquals(rs.getShort(1), 1);
+                    assertTrue(rs.next());
+                    assertEquals(rs.getShort(1), 2);
+                    assertTrue(rs.next());
+                    assertEquals(rs.getShort(1), 3);
+                    assertTrue(rs.next());
+                    assertEquals(rs.getShort(1), 4);
+                    assertFalse(rs.next());
+                }
+            }
+        }
+    }
+
     @Test(groups = {"integration"})
     public void testJdbcEscapeSyntax() throws Exception {
         if (ClickHouseVersion.of(getServerVersion()).check("(,23.8]")) {