[jira] [Created] (FLINK-36930) Lookup join with JDBC connector fails when join condition does not include fields from the probe/left table

[Lilo (Jira)]

Lilo created FLINK-36930:
----------------------------

             Summary: Lookup join with JDBC connector fails when join condition 
does not include fields from the probe/left table
                 Key: FLINK-36930
                 URL: https://issues.apache.org/jira/browse/FLINK-36930
             Project: Flink
          Issue Type: Bug
    Affects Versions: 1.19.1, 1.18.1, 1.17.0
         Environment:  * Flink versions affected: 1.17.0 and later versions 
(Tested on 1.19.1)
 * Flink version where this worked: 1.16.2
 * Connector: JDBC (tested with Apache Derby and MySQL ; likely reproducible 
with other JDBC drivers). Notably, this issue does not affect the Hive 
connector.

            Reporter: Lilo


I've encountered a regression in Flink 1.17.0 (and later versions) related to 
lookup joins when using the JDBC connector as the lookup source. The issue 
arises when the join condition does not include fields from the probe/left 
table. This scenario worked correctly in Flink 1.16.2 but now throws a 
`TableException`.
 

 
The following SQL code demonstrates the problem.
{code:sql}
SET 'execution.target' = 'local';

-- Create the lookup table (using Apache Derby as an example)
DROP TEMPORARY TABLE IF EXISTS lookup_table;
CREATE TEMPORARY TABLE lookup_table (
    id INT,
    sub_id INT,
    v STRING
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:derby:memory:myInMemoryDB;create=true',
    'table-name' = 'lookup_table'
);

-- Create the main table (using datagen connector)
DROP TEMPORARY TABLE IF EXISTS main_table;
CREATE TEMPORARY TABLE main_table (
    v STRING,
    some_id INT,
    some_sub_id INT,
    proctime AS PROCTIME()
) WITH (
    'connector' = 'datagen',
    'rows-per-second' = '2',
    'fields.some_id.kind' = 'random',
    'fields.some_id.min' = '1',
    'fields.some_id.max' = '5',
    'fields.some_sub_id.kind' = 'random',
    'fields.some_sub_id.min' = '1',
    'fields.some_sub_id.max' = '3',
    'fields.v.length' = '10'
);

-- This lookup join works correctly (join condition includes a field from 
main_table)
EXPLAIN PLAN FOR
SELECT
    t1.*,
    t2.v AS lookup_value
FROM main_table t1
INNER JOIN lookup_table FOR SYSTEM_TIME AS OF t1.proctime AS t2
    ON t2.id = 1
    AND t2.sub_id = t1.some_sub_id;

-- This lookup join FAILS in Flink 1.17.0+ (join condition only uses constants)
EXPLAIN PLAN FOR
SELECT
    t1.*,
    t2.v AS lookup_value
FROM main_table t1
INNER JOIN lookup_table FOR SYSTEM_TIME AS OF t1.proctime AS t2
    ON t2.id = 1
    AND t2.sub_id = 1;

-- Attempting to bypass the issue with a view also fails
DROP VIEW IF EXISTS main_table_view;
CREATE TEMPORARY VIEW main_table_view AS SELECT *, 1 AS fake_id FROM main_table;
EXPLAIN PLAN FOR
SELECT
    t1.*,
    t2.v AS lookup_value
FROM main_table_view t1
INNER JOIN lookup_table FOR SYSTEM_TIME AS OF t1.proctime AS t2
    ON t2.id = 1
    AND t2.sub_id = t1.fake_id;
{code}

*Expected Behavior:*

These queries should execute successfully, even when the join condition does 
not involve fields from the main table. This was the behavior in Flink 1.16.2.

*Actual Behavior:*

In Flink 1.17.0 and later, the second query (where the join condition only uses 
constants) throws the following exception (from Flink 1.17.2):

{code:java}
Caused by: org.apache.flink.table.api.TableException: Temporal table join 
requires an equality condition on fields of table 
[default_catalog.default_database.lookup_table].
        at 
org.apache.flink.table.planner.plan.nodes.exec.common.CommonExecLookupJoin.validate(CommonExecLookupJoin.java:687)
        at 
org.apache.flink.table.planner.plan.nodes.exec.common.CommonExecLookupJoin.createJoinTransformation(CommonExecLookupJoin.java:249)
        at 
org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecLookupJoin.translateToPlanInternal(StreamExecLookupJoin.java:157)
        at 
org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase.translateToPlan(ExecNodeBase.java:161)
        at 
org.apache.flink.table.planner.plan.nodes.exec.ExecEdge.translateToPlan(ExecEdge.java:257)
        at 
org.apache.flink.table.planner.plan.nodes.exec.common.CommonExecCalc.translateToPlanInternal(CommonExecCalc.java:94)
        at 
org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase.translateToPlan(ExecNodeBase.java:161)
        at 
org.apache.flink.table.planner.delegation.StreamPlanner.$anonfun$translateToPlan$1(StreamPlanner.scala:85)
{code}




--
This message was sent by Atlassian Jira
(v8.20.10#820010)