[improvement](jdbc catalog) Disallow non-constant type conversion pushdown and implicit conversion pushdown #42102

docker/thirdparties/docker-compose/mysql/init/03-create-table.sql

@@ -505,3 +505,10 @@ CREATE TABLE doris_test.`t_char` (
   `char_col` char(255) COLLATE utf8_bin DEFAULT NULL
 );
 
+CREATE TABLE doris_test.`test_cast` (
+  `id` int(11) DEFAULT NULL,
+  `int_c` varchar(100),
+  `date_c` varchar(100),
+  `datetime_c` varchar(100)
+);
+

docker/thirdparties/docker-compose/mysql/init/04-insert.sql

@@ -1206,3 +1206,5 @@ INSERT INTO doris_test.t_char (char_col) VALUES (REPEAT('a', 255));
 
 INSERT INTO doris_test.t_char (char_col) VALUES (REPEAT('中', 255));
 
+INSERT INTO doris_test.`test_cast` VALUES (1, '1', '2022-01-01', '2022-01-01 00:00:01');
+INSERT INTO doris_test.`test_cast` VALUES (2, '2', '2022-01-02', '2022-01-02 00:00:01');

fe/fe-core/src/main/java/org/apache/doris/datasource/jdbc/source/JdbcScanNode.java

@@ -20,6 +20,7 @@
 import org.apache.doris.analysis.Analyzer;
 import org.apache.doris.analysis.BinaryPredicate;
 import org.apache.doris.analysis.BoolLiteral;
+import org.apache.doris.analysis.CastExpr;
 import org.apache.doris.analysis.DateLiteral;
 import org.apache.doris.analysis.Expr;
 import org.apache.doris.analysis.ExprSubstitutionMap;
@@ -308,11 +309,20 @@ public StatsDelta genStatsDelta() throws AnalysisException {
     private static boolean shouldPushDownConjunct(TOdbcTableType tableType, Expr expr) {
         // Prevent pushing down expressions with NullLiteral to Oracle
         if (ConnectContext.get() != null
-                && !ConnectContext.get().getSessionVariable().jdbcOracleNullPredicatePushdown
+                && !ConnectContext.get().getSessionVariable().enableJdbcOracleNullPredicatePushDown
                 && containsNullLiteral(expr)
                 && tableType.equals(TOdbcTableType.ORACLE)) {
             return false;
         }
+
+        // Prevent pushing down cast expressions if ConnectContext is null or cast pushdown is disabled
+        if (ConnectContext.get() == null || !ConnectContext.get()
+                .getSessionVariable().enableJdbcCastPredicatePushDown) {
+            if (containsCastExpr(expr)) {
+                return false;
+            }
+        }
+
         if (containsFunctionCallExpr(expr)) {
             if (tableType.equals(TOdbcTableType.MYSQL) || tableType.equals(TOdbcTableType.CLICKHOUSE)
                     || tableType.equals(TOdbcTableType.ORACLE)) {
@@ -384,4 +394,10 @@ private static boolean containsNullLiteral(Expr expr) {
         expr.collect(NullLiteral.class, nullExprList);
         return !nullExprList.isEmpty();
     }
+
+    private static boolean containsCastExpr(Expr expr) {
+        List<CastExpr> castExprList = Lists.newArrayList();
+        expr.collect(CastExpr.class, castExprList);
+        return !castExprList.isEmpty();
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java

@@ -504,7 +504,10 @@ public class SessionVariable implements Serializable, Writable {
 
     public static final String JDBC_CLICKHOUSE_QUERY_FINAL = "jdbc_clickhouse_query_final";
 
-    public static final String JDBC_ORACLE_NULL_PREDICATE_PUSHDOWN = "jdbc_oracle_null_predicate_pushdown";
+    public static final String ENABLE_JDBC_ORACLE_NULL_PREDICATE_PUSH_DOWN
+            = "enable_jdbc_oracle_null_predicate_push_down";
+
+    public static final String ENABLE_JDBC_CAST_PREDICATE_PUSH_DOWN = "enable_jdbc_cast_predicate_push_down";
 
     public static final String ENABLE_MEMTABLE_ON_SINK_NODE =
             "enable_memtable_on_sink_node";
@@ -687,10 +690,15 @@ public class SessionVariable implements Serializable, Writable {
                     "Whether to add the FINAL keyword to the query SQL when querying ClickHouse JDBC external tables."})
     public boolean jdbcClickhouseQueryFinal = false;
 
-    @VariableMgr.VarAttr(name = JDBC_ORACLE_NULL_PREDICATE_PUSHDOWN, needForward = true,
+    @VariableMgr.VarAttr(name = ENABLE_JDBC_ORACLE_NULL_PREDICATE_PUSH_DOWN, needForward = true,
             description = {"是否允许将 NULL 谓词下推到 Oracle JDBC 外部表。",
                     "Whether to allow NULL predicates to be pushed down to Oracle JDBC external tables."})
-    public boolean jdbcOracleNullPredicatePushdown = false;
+    public boolean enableJdbcOracleNullPredicatePushDown = false;
+
+    @VariableMgr.VarAttr(name = ENABLE_JDBC_CAST_PREDICATE_PUSH_DOWN, needForward = true,
+            description = {"是否允许将带有 CAST 表达式的谓词下推到 JDBC 外部表。",
+                    "Whether to allow predicates with CAST expressions to be pushed down to JDBC external tables."})
+    public boolean enableJdbcCastPredicatePushDown = false;
 
     @VariableMgr.VarAttr(name = ROUND_PRECISE_DECIMALV2_VALUE)
     public boolean roundPreciseDecimalV2Value = false;

regression-test/data/external_table_p0/jdbc/test_jdbc_catalog_push_cast.out

@@ -0,0 +1,4 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !sql --
+1	1	2022-01-01	2022-01-01 00:00:01
+

regression-test/suites/external_table_p0/jdbc/test_jdbc_catalog_push_cast.groovy

@@ -0,0 +1,105 @@
+// 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.
+
+suite("test_jdbc_catalog_push_cast", "p0,external,mysql,external_docker,external_docker_mysql") {
+    String enabled = context.config.otherConfigs.get("enableJdbcTest")
+    String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+    String s3_endpoint = getS3Endpoint()
+    String bucket = getS3BucketName()
+    String driver_url = "https://${bucket}.${s3_endpoint}/regression/jdbc_driver/mysql-connector-j-8.3.0.jar"
+    if (enabled != null && enabled.equalsIgnoreCase("true")) {
+        String mysql_port = context.config.otherConfigs.get("mysql_57_port");
+
+        sql """drop catalog if exists jdbc_catalog_push_cast """
+        sql """create catalog if not exists jdbc_catalog_push_cast properties(
+            "type"="jdbc",
+            "user"="root",
+            "password"="123456",
+            "jdbc_url" = "jdbc:mysql://${externalEnvIp}:${mysql_port}/doris_test?useSSL=false",
+            "driver_url" = "${driver_url}",
+            "driver_class" = "com.mysql.cj.jdbc.Driver"
+        );"""
+
+        sql "use jdbc_catalog_push_cast.doris_test"
+
+        qt_sql """select * from test_cast where date(datetime_c) = '2022-01-01';"""
+
+        explain {
+            sql("select * from test_cast where date(datetime_c) = '2022-01-01';")
+            contains("QUERY: SELECT `id`, `int_c`, `date_c`, `datetime_c` FROM `doris_test`.`test_cast`")
+        }
+
+        explain {
+            sql("select * from test_cast where datetime_c = now();")
+            contains("QUERY: SELECT `id`, `int_c`, `date_c`, `datetime_c` FROM `doris_test`.`test_cast`")
+        }
+
+        explain {
+            sql("select * from test_cast where datetime_c = cast(cast('2022-01-01 00:00:01' as datetime) as string);")
+            contains("QUERY: SELECT `id`, `int_c`, `date_c`, `datetime_c` FROM `doris_test`.`test_cast` WHERE ((`datetime_c` = '2022-01-01 00:00:01'))")
+        }
+
+        explain {
+            sql("select * from test_cast where cast(datetime_c as datetime) = cast('2022-01-01 00:00:01' as datetime);")
+            contains("QUERY: SELECT `id`, `int_c`, `date_c`, `datetime_c` FROM `doris_test`.`test_cast`")
+        }
+
+        explain {
+            sql("select * from test_cast where date_c = cast(cast('2022-01-01 00:00:01' as datetime) as date)")
+            contains("QUERY: SELECT `id`, `int_c`, `date_c`, `datetime_c` FROM `doris_test`.`test_cast`")
+        }
+
+        explain {
+            sql("select * from test_cast where datetime_c = date_c;")
+            contains("QUERY: SELECT `id`, `int_c`, `date_c`, `datetime_c` FROM `doris_test`.`test_cast`")
+        }
+
+        explain {
+            sql("select * from test_cast where datetime_c = '2022-01-01';")
+            contains("QUERY: SELECT `id`, `int_c`, `date_c`, `datetime_c` FROM `doris_test`.`test_cast`")
+        }
+
+        explain {
+            sql("select * from test_cast where cast(datetime_c as string) = '2022-01-01 00:00:01';")
+            contains("SELECT `id`, `int_c`, `date_c`, `datetime_c` FROM `doris_test`.`test_cast`")
+        }
+
+        explain {
+            sql("select * from test_cast where datetime_c != '2022-01-01 00:00:01';")
+            contains("QUERY: SELECT `id`, `int_c`, `date_c`, `datetime_c` FROM `doris_test`.`test_cast` WHERE ((`datetime_c` != '2022-01-01 00:00:01'))")
+        }
+
+        explain {
+            sql("select * from test_cast where datetime_c in (cast('2022-01-01' as datetime), cast('2022-02-01' as datetime));")
+            contains("QUERY: SELECT `id`, `int_c`, `date_c`, `datetime_c` FROM `doris_test`.`test_cast`")
+        }
+
+        explain {
+            sql("select * from test_cast where date_c = cast(cast('2022-01-01 00:00:01' as datetime) as date) and id = 1;")
+            contains("QUERY: SELECT `id`, `int_c`, `date_c`, `datetime_c` FROM `doris_test`.`test_cast` WHERE ((`id` = 1))")
+        }
+
+        sql "set enable_jdbc_cast_predicate_push_down = true;"
+
+        explain {
+            sql("select * from test_cast where cast(datetime_c as datetime) = cast('2022-01-01 00:00:01' as datetime);")
+            contains("QUERY: SELECT `id`, `int_c`, `date_c`, `datetime_c` FROM `doris_test`.`test_cast` WHERE (`datetime_c` = '2022-01-01 00:00:01')")
+        }
+
+        sql """drop catalog if exists jdbc_catalog_push_cast """
+    }
+}