This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-jdbc.git
The following commit(s) were added to refs/heads/main by this push:
new 97914f47 [FLINK-40778][tests] Handle Flink 2.4's AND digest in
JdbcTablePlanTest
97914f47 is described below
commit 97914f47c458b5215d176e7e31af04c5ceccfa4a
Author: Martijn Visser <[email protected]>
AuthorDate: Mon Oct 5 10:09:07 2026 +0200
[FLINK-40778][tests] Handle Flink 2.4's AND digest in JdbcTablePlanTest
FLINK-40021 changed the pushed-down filter digest from and(...) to
AND(...) in Flink 2.4. Main still builds against 2.1 to 2.3, so keep
the existing reference plan and add one for 2.4 and later.
Generated-by: Claude Code (Claude Opus 5.5)
Co-authored-by: Purushottam Sinha <[email protected]>
---
.../connector/jdbc/core/table/JdbcTablePlanTest.java | 20 ++++++++++++++++++++
.../connector/jdbc/core/table/JdbcTablePlanTest.xml | 17 +++++++++++++++++
2 files changed, 37 insertions(+)
diff --git
a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/table/JdbcTablePlanTest.java
b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/table/JdbcTablePlanTest.java
index 0e587009..5d7b9187 100644
---
a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/table/JdbcTablePlanTest.java
+++
b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/table/JdbcTablePlanTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.connector.jdbc.core.table;
+import org.apache.flink.FlinkVersion;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.planner.utils.StreamTableTestUtil;
import org.apache.flink.table.planner.utils.TableTestBase;
@@ -27,9 +28,14 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInfo;
import org.junit.rules.TestName;
+import static org.assertj.core.api.Assumptions.assumeThat;
+
/** Plan tests for JDBC connector, for example, testing projection push down.
*/
class JdbcTablePlanTest extends TableTestBase {
+ /** Flink 2.4 spells the AND of pushed-down filters in upper case
(FLINK-40021). */
+ private static final boolean UPPER_CASE_FILTER_AND =
FlinkVersion.byCode("2.4").isPresent();
+
private final StreamTableTestUtil util =
streamTestUtil(TableConfig.getDefault());
private TestInfo testInfo;
@@ -95,8 +101,22 @@ class JdbcTablePlanTest extends TableTestBase {
util.verifyExecPlan("SELECT id, time_col FROM jdbc LIMIT 3");
}
+ /**
+ * Remove this test, its plan and UPPER_CASE_FILTER_AND once Flink < 2.4
is no longer supported.
+ */
@Test
void testFilterPushdown() {
+ assumeThat(UPPER_CASE_FILTER_AND).isFalse();
+ verifyFilterPushdown();
+ }
+
+ @Test
+ void testFilterPushdownSinceFlink24() {
+ assumeThat(UPPER_CASE_FILTER_AND).isTrue();
+ verifyFilterPushdown();
+ }
+
+ private void verifyFilterPushdown() {
util.verifyExecPlan(
"SELECT id, time_col, real_col FROM jdbc WHERE id = 900001 AND
time_col <> TIME '11:11:11' OR double_col >= -1000.23");
}
diff --git
a/flink-connector-jdbc-core/src/test/resources/org/apache/flink/connector/jdbc/core/table/JdbcTablePlanTest.xml
b/flink-connector-jdbc-core/src/test/resources/org/apache/flink/connector/jdbc/core/table/JdbcTablePlanTest.xml
index d24b3891..c461382b 100644
---
a/flink-connector-jdbc-core/src/test/resources/org/apache/flink/connector/jdbc/core/table/JdbcTablePlanTest.xml
+++
b/flink-connector-jdbc-core/src/test/resources/org/apache/flink/connector/jdbc/core/table/JdbcTablePlanTest.xml
@@ -30,6 +30,23 @@ LogicalProject(id=[$0], time_col=[$3], real_col=[$4])
<Resource name="optimized exec plan">
<![CDATA[
TableSourceScan(table=[[default_catalog, default_database, jdbc,
filter=[and(OR(=(id, 900001:BIGINT), >=(double_col, -1000.23:DECIMAL(6, 2))),
OR(<>(time_col, 11:11:11), >=(double_col, -1000.23:DECIMAL(6, 2))))],
project=[id, time_col, real_col]]], fields=[id, time_col, real_col])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testFilterPushdownSinceFlink24">
+ <Resource name="sql">
+ <![CDATA[SELECT id, time_col, real_col FROM jdbc WHERE id = 900001
AND time_col <> TIME '11:11:11' OR double_col >= -1000.23]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(id=[$0], time_col=[$3], real_col=[$4])
++- LogicalFilter(condition=[OR(AND(=($0, 900001), <>($3, 11:11:11)), >=($5,
-1000.23:DECIMAL(6, 2)))])
+ +- LogicalTableScan(table=[[default_catalog, default_database, jdbc]])
+]]>
+ </Resource>
+ <Resource name="optimized exec plan">
+ <![CDATA[
+TableSourceScan(table=[[default_catalog, default_database, jdbc,
filter=[AND(OR(=(id, 900001:BIGINT), >=(double_col, -1000.23:DECIMAL(6, 2))),
OR(<>(time_col, 11:11:11), >=(double_col, -1000.23:DECIMAL(6, 2))))],
project=[id, time_col, real_col]]], fields=[id, time_col, real_col])
]]>
</Resource>
</TestCase>
