Repository: clickhouse/clickhouse-java
Issue: Question about emptyBatchError

Currently, I use Flink JDBC + CK Driver to write data to clickhouse in real time, and there is a high probability of encountering the following errors , So maybe we can support writing empty batch ? 
```
Caused by: java.lang.RuntimeException: Writing records to JDBC failed.
	at org.apache.flink.connector.jdbc.internal.JdbcBatchingOutputFormat.checkFlushException(JdbcBatchingOutputFormat.java:154)
	at org.apache.flink.connector.jdbc.internal.JdbcBatchingOutputFormat.writeRecord(JdbcBatchingOutputFormat.java:160)
	at org.apache.flink.streaming.api.functions.sink.OutputFormatSinkFunction.invoke(OutputFormatSinkFunction.java:87)
	at org.apache.flink.streaming.api.functions.sink.SinkFunction.invoke(SinkFunction.java:49)
	at org.apache.flink.table.runtime.operators.sink.SinkOperator.processElement(SinkOperator.java:73)
	at org.apache.flink.streaming.runtime.tasks.ChainingOutput.pushToOperator(ChainingOutput.java:113)
	at org.apache.flink.streaming.runtime.tasks.ChainingOutput.collect(ChainingOutput.java:94)
	at org.apache.flink.streaming.runtime.tasks.ChainingOutput.collect(ChainingOutput.java:40)
	at org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask$AsyncDataOutputToOutput.emitRecord(SourceOperatorStreamTask.java:163)
	at org.apache.flink.streaming.api.operators.source.SourceOutputWithWatermarks.collect(SourceOutputWithWatermarks.java:110)
	at org.apache.flink.streaming.api.operators.source.SourceOutputWithWatermarks.collect(SourceOutputWithWatermarks.java:101)
	at org.apache.flink.connector.file.src.impl.FileSourceRecordEmitter.emitRecord(FileSourceRecordEmitter.java:45)
	at org.apache.flink.connector.file.src.impl.FileSourceRecordEmitter.emitRecord(FileSourceRecordEmitter.java:35)
	at org.apache.flink.connector.base.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:128)
	at org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:275)
	at org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:67)
	at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:398)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:191)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:619)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:583)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:758)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:573)
	at java.lang.Thread.run(Thread.java:748)
Caused by: java.io.IOException: java.sql.SQLException: Please call addBatch method at least once before batch execution
	at org.apache.flink.connector.jdbc.internal.JdbcBatchingOutputFormat.flush(JdbcBatchingOutputFormat.java:190)
	at org.apache.flink.connector.jdbc.internal.JdbcBatchingOutputFormat.lambda$open$0(JdbcBatchingOutputFormat.java:128)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	... 1 more
Caused by: java.sql.SQLException: Please call addBatch method at least once before batch execution
	at com.clickhouse.jdbc.SqlExceptionUtils.clientError(SqlExceptionUtils.java:43)
	at com.clickhouse.jdbc.SqlExceptionUtils.emptyBatchError(SqlExceptionUtils.java:111)
	at com.clickhouse.jdbc.internal.InputBasedPreparedStatement.executeAny(InputBasedPreparedStatement.java:95)
	at com.clickhouse.jdbc.internal.AbstractPreparedStatement.executeLargeBatch(AbstractPreparedStatement.java:85)
	at com.clickhouse.jdbc.internal.ClickHouseStatementImpl.executeBatch(ClickHouseStatementImpl.java:568)
	at org.apache.flink.connector.jdbc.statement.FieldNamedPreparedStatementImpl.executeBatch(FieldNamedPreparedStatementImpl.java:65)
	at org.apache.flink.connector.jdbc.internal.executor.TableSimpleStatementExecutor.executeBatch(TableSimpleStatementExecutor.java:64)
	at org.apache.flink.connector.jdbc.internal.executor.TableBufferedStatementExecutor.executeBatch(TableBufferedStatementExecutor.java:64)
	at org.apache.flink.connector.jdbc.internal.JdbcBatchingOutputFormat.attemptFlush(JdbcBatchingOutputFormat.java:216)
	at org.apache.flink.connector.jdbc.internal.JdbcBatchingOutputFormat.flush(JdbcBatchingOutputFormat.java:184)
	... 8 more

```

Changed files:
clickhouse-jdbc/src/test/java/com/clickhouse/jdbc/ClickHouseStatementTest.java
clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/internal/SqlBasedPreparedStatement.java
clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/internal/ClickHouseConnectionImpl.java
clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/SqlExceptionUtils.java
clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/internal/InputBasedPreparedStatement.java
clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/internal/ClickHouseStatementImpl.java
clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/internal/TableBasedPreparedStatement.java
clickhouse-jdbc/src/test/java/com/clickhouse/jdbc/ClickHousePreparedStatementTest.java

diff --git a/clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/SqlExceptionUtils.java b/clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/SqlExceptionUtils.java
index da5b824b..25b3ffe3 100644
--- a/clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/SqlExceptionUtils.java
+++ b/clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/SqlExceptionUtils.java
@@ -107,10 +107,6 @@ public final class SqlExceptionUtils {
         return new BatchUpdateException("Unexpected error", SQL_STATE_SQL_ERROR, 0, updateCounts, cause);
     }
 
-    public static SQLException emptyBatchError() {
-        return clientError("Please call addBatch method at least once before batch execution");
-    }
-
     public static BatchUpdateException queryInBatchError(int[] updateCounts) {
         return new BatchUpdateException("Query is not allow in batch update", SQL_STATE_CLIENT_ERROR, updateCounts);
     }
diff --git a/clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/internal/ClickHouseConnectionImpl.java b/clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/internal/ClickHouseConnectionImpl.java
index bae6de91..be75aa33 100644
--- a/clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/internal/ClickHouseConnectionImpl.java
+++ b/clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/internal/ClickHouseConnectionImpl.java
@@ -220,7 +220,7 @@ public class ClickHouseConnectionImpl extends JdbcWrapper implements ClickHouseC
         autoCommit = !jdbcConf.isJdbcCompliant() || jdbcConf.isAutoCommit();
 
         ClickHouseNode node = connInfo.getServer();
-        log.debug("Connecting to node: %s", node);
+        log.debug("Connecting to: %s", node);
 
         jvmTimeZone = TimeZone.getDefault();
 
diff --git a/clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/internal/ClickHouseStatementImpl.java b/clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/internal/ClickHouseStatementImpl.java
index 2b64d85c..94de8e5e 100644
--- a/clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/internal/ClickHouseStatementImpl.java
+++ b/clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/internal/ClickHouseStatementImpl.java
@@ -25,6 +25,7 @@ import com.clickhouse.client.ClickHouseResponseSummary;
 import com.clickhouse.client.ClickHouseSerializer;
 import com.clickhouse.client.ClickHouseUtils;
 import com.clickhouse.client.ClickHouseValue;
+import com.clickhouse.client.ClickHouseValues;
 import com.clickhouse.client.config.ClickHouseClientOption;
 import com.clickhouse.client.config.ClickHouseConfigChangeListener;
 import com.clickhouse.client.config.ClickHouseOption;
@@ -588,7 +589,7 @@ public class ClickHouseStatementImpl extends JdbcWrapper
     public long[] executeLargeBatch() throws SQLException {
         ensureOpen();
         if (batchStmts.isEmpty()) {
-            throw SqlExceptionUtils.emptyBatchError();
+            return ClickHouseValues.EMPTY_LONG_ARRAY;
         }
 
         boolean continueOnError = getConnection().getJdbcConfig().isContinueBatchOnError();
diff --git a/clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/internal/InputBasedPreparedStatement.java b/clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/internal/InputBasedPreparedStatement.java
index 28c09b80..fd676de9 100644
--- a/clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/internal/InputBasedPreparedStatement.java
+++ b/clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/internal/InputBasedPreparedStatement.java
@@ -94,7 +94,7 @@ public class InputBasedPreparedStatement extends AbstractPreparedStatement imple
         boolean continueOnError = false;
         if (asBatch) {
             if (counter < 1) {
-                throw SqlExceptionUtils.emptyBatchError();
+                return ClickHouseValues.EMPTY_LONG_ARRAY;
             }
             continueOnError = getConnection().getJdbcConfig().isContinueBatchOnError();
         } else {
diff --git a/clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/internal/SqlBasedPreparedStatement.java b/clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/internal/SqlBasedPreparedStatement.java
index 8958f95d..664f999e 100644
--- a/clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/internal/SqlBasedPreparedStatement.java
+++ b/clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/internal/SqlBasedPreparedStatement.java
@@ -122,7 +122,7 @@ public class SqlBasedPreparedStatement extends AbstractPreparedStatement impleme
         boolean continueOnError = false;
         if (asBatch) {
             if (counter < 1) {
-                throw SqlExceptionUtils.emptyBatchError();
+                return ClickHouseValues.EMPTY_LONG_ARRAY;
             }
             continueOnError = getConnection().getJdbcConfig().isContinueBatchOnError();
         } else {
diff --git a/clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/internal/TableBasedPreparedStatement.java b/clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/internal/TableBasedPreparedStatement.java
index c389b589..e86c94e2 100644
--- a/clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/internal/TableBasedPreparedStatement.java
+++ b/clickhouse-jdbc/src/main/java/com/clickhouse/jdbc/internal/TableBasedPreparedStatement.java
@@ -19,6 +19,7 @@ import java.util.Set;
 import com.clickhouse.client.ClickHouseRequest;
 import com.clickhouse.client.ClickHouseResponse;
 import com.clickhouse.client.ClickHouseUtils;
+import com.clickhouse.client.ClickHouseValues;
 import com.clickhouse.client.data.ClickHouseExternalTable;
 import com.clickhouse.client.logging.Logger;
 import com.clickhouse.client.logging.LoggerFactory;
@@ -75,7 +76,7 @@ public class TableBasedPreparedStatement extends AbstractPreparedStatement imple
         boolean continueOnError = false;
         if (asBatch) {
             if (batch.isEmpty()) {
-                throw SqlExceptionUtils.emptyBatchError();
+                return ClickHouseValues.EMPTY_LONG_ARRAY;
             }
             continueOnError = getConnection().getJdbcConfig().isContinueBatchOnError();
         } else {
diff --git a/clickhouse-jdbc/src/test/java/com/clickhouse/jdbc/ClickHousePreparedStatementTest.java b/clickhouse-jdbc/src/test/java/com/clickhouse/jdbc/ClickHousePreparedStatementTest.java
index 2a811b79..c131e052 100644
--- a/clickhouse-jdbc/src/test/java/com/clickhouse/jdbc/ClickHousePreparedStatementTest.java
+++ b/clickhouse-jdbc/src/test/java/com/clickhouse/jdbc/ClickHousePreparedStatementTest.java
@@ -778,15 +778,22 @@ public class ClickHousePreparedStatementTest extends JdbcIntegrationTest {
     public void testBatchQuery() throws SQLException {
         try (ClickHouseConnection conn = newConnection(new Properties());
                 PreparedStatement stmt = conn.prepareStatement("select * from numbers(100) where number < ?")) {
+            Assert.assertEquals(stmt.executeBatch(), new int[0]);
+            Assert.assertEquals(stmt.executeLargeBatch(), new long[0]);
             Assert.assertThrows(SQLException.class, () -> stmt.setInt(0, 5));
             Assert.assertThrows(SQLException.class, () -> stmt.setInt(2, 5));
             Assert.assertThrows(SQLException.class, () -> stmt.addBatch());
 
             stmt.setInt(1, 3);
+            Assert.assertEquals(stmt.executeBatch(), new int[0]);
+            Assert.assertEquals(stmt.executeLargeBatch(), new long[0]);
             stmt.addBatch();
             stmt.setInt(1, 2);
             stmt.addBatch();
             Assert.assertThrows(BatchUpdateException.class, () -> stmt.executeBatch());
+
+            Assert.assertEquals(stmt.executeBatch(), new int[0]);
+            Assert.assertEquals(stmt.executeLargeBatch(), new long[0]);
         }
     }
 
diff --git a/clickhouse-jdbc/src/test/java/com/clickhouse/jdbc/ClickHouseStatementTest.java b/clickhouse-jdbc/src/test/java/com/clickhouse/jdbc/ClickHouseStatementTest.java
index 9a61a4bf..f0847918 100644
--- a/clickhouse-jdbc/src/test/java/com/clickhouse/jdbc/ClickHouseStatementTest.java
+++ b/clickhouse-jdbc/src/test/java/com/clickhouse/jdbc/ClickHouseStatementTest.java
@@ -136,6 +136,9 @@ public class ClickHouseStatementTest extends JdbcIntegrationTest {
     public void testMutation() throws SQLException {
         Properties props = new Properties();
         try (ClickHouseConnection conn = newConnection(props); ClickHouseStatement stmt = conn.createStatement()) {
+            Assert.assertEquals(stmt.executeBatch(), new int[0]);
+            Assert.assertEquals(stmt.executeLargeBatch(), new long[0]);
+
             Assert.assertFalse(stmt.execute("drop table if exists test_mutation;"
                     + "create table test_mutation(a String, b UInt32) engine=MergeTree() order by tuple()"),
                     "Should not return result set");
@@ -156,6 +159,9 @@ public class ClickHouseStatementTest extends JdbcIntegrationTest {
             stmt.addBatch("drop table non_existing_table");
             stmt.addBatch("insert into test_mutation values('2',2)");
             Assert.assertThrows(SQLException.class, () -> stmt.executeBatch());
+
+            Assert.assertEquals(stmt.executeBatch(), new int[0]);
+            Assert.assertEquals(stmt.executeLargeBatch(), new long[0]);
         }
 
         props.setProperty(JdbcConfig.PROP_CONTINUE_BATCH, "true");
@@ -269,7 +275,12 @@ public class ClickHouseStatementTest extends JdbcIntegrationTest {
     public void testExecuteBatch() throws SQLException {
         Properties props = new Properties();
         try (Connection conn = newConnection(props); Statement stmt = conn.createStatement()) {
-            Assert.assertThrows(SQLException.class, () -> stmt.executeBatch());
+            Assert.assertEquals(stmt.executeBatch(), new int[0]);
+            Assert.assertEquals(stmt.executeLargeBatch(), new long[0]);
+            stmt.addBatch("select 1");
+            stmt.clearBatch();
+            Assert.assertEquals(stmt.executeBatch(), new int[0]);
+            Assert.assertEquals(stmt.executeLargeBatch(), new long[0]);
             stmt.addBatch("select 1");
             // mixed usage
             Assert.assertThrows(SQLException.class, () -> stmt.execute("select 2"));