[Fix][Connector-v2][Clickhouse]Fix filter_query not working in SQL batch read strategy

seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/ClickhouseSinkFactory.java

@@ -20,6 +20,7 @@
 import org.apache.seatunnel.api.configuration.ReadonlyConfig;
 import org.apache.seatunnel.api.configuration.util.OptionRule;
 import org.apache.seatunnel.api.options.SinkConnectorCommonOptions;
+import org.apache.seatunnel.api.sink.DataSaveMode;
 import org.apache.seatunnel.api.table.catalog.CatalogTable;
 import org.apache.seatunnel.api.table.connector.TableSink;
 import org.apache.seatunnel.api.table.factory.Factory;
@@ -75,9 +76,9 @@ public OptionRule optionRule() {
                         ALLOW_EXPERIMENTAL_LIGHTWEIGHT_DELETE,
                         SCHEMA_SAVE_MODE,
                         DATA_SAVE_MODE,
-                        CUSTOM_SQL,
                         SAVE_MODE_CREATE_TEMPLATE,
                         SinkConnectorCommonOptions.MULTI_TABLE_SINK_REPLICA)
+                .conditional(DATA_SAVE_MODE, DataSaveMode.CUSTOM_PROCESSING, CUSTOM_SQL)
                 .build();
     }
 }

seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/source/ClickhouseValueReader.java

@@ -306,6 +306,15 @@ private String buildBatchSqlQuery() {
             }
         }
 
+        // Add filter_query support for SQL batch strategy
+        if (StringUtils.isNotEmpty(clickhouseSourceTable.getFilterQuery())) {
+            if (whereClause.isEmpty()) {
+                whereClause = " WHERE (" + clickhouseSourceTable.getFilterQuery() + ")";
+            } else {
+                whereClause += " AND (" + clickhouseSourceTable.getFilterQuery() + ")";
+            }
+        }
+
         String orderByClause = " ORDER BY " + sortingKey;
 
         String sql;

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-clickhouse-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/ClickhouseIT.java

@@ -262,6 +262,18 @@ public void testClickhouseWitJoinComplexSql(TestContainer testContainer)
         clearTable(SINK_TABLE);
     }
 
+    @TestTemplate
+    public void testClickhouseWithSqlAndFilterQuery(TestContainer testContainer)
+            throws IOException, InterruptedException {
+        Container.ExecResult execResult =
+                testContainer.executeJob("/clickhouse_with_sql_and_filter_query.conf");
+        Assertions.assertEquals(0, execResult.getExitCode());
+        Assertions.assertEquals(100, countData(SOURCE_MERGE_TREE_TABLE));
+        // filter_query = "id < 47" should filter data to 47 rows (id from 0 to 46)
+        Assertions.assertEquals(47, countData(SINK_TABLE));
+        clearTable(SINK_TABLE);
+    }
+
     @TestTemplate
     public void testClickhouseWithMultiTableSource(TestContainer testContainer)
             throws IOException, InterruptedException {

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-clickhouse-e2e/src/test/resources/clickhouse_with_sql_and_filter_query.conf

@@ -0,0 +1,53 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+######
+###### This config file tests filter_query with SQL batch strategy
+######
+
+env {
+  parallelism = 3
+  job.mode = "BATCH"
+}
+
+source {
+  # Test filter_query support in SQL batch strategy
+  Clickhouse {
+    host = "clickhouse:8123"
+    username = "default"
+    password = ""
+    table_path = "default.source_merge_tree_table"
+    sql = "select * from default.source_merge_tree_table"
+    filter_query = "id < 47"
+    batch_size = 10
+  }
+  # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
+  # please go to https://seatunnel.apache.org/docs/connector-v2/source/ClickhouseSource
+}
+
+sink {
+  Clickhouse {
+    host = "clickhouse:8123"
+    database = "default"
+    table = "sink_table"
+    username = "default"
+    password = ""
+  }
+
+  # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
+  # please go to https://seatunnel.apache.org/docs/connector-v2/sink
+}
+