[FLINK-36808][table-planner] Fix LookupJoin bug when used with filterable table source

[morazow]

What is the purpose of the change

• Fix bug in LookupJoin that occurs when used together with filterable (that support filter pushdown) table and union all

Brief description of the bug

Given a union all query:

• that combines results of the two lookup joins
• that these lookup joins have different filter queries

SELECT
     s.id,
     s.name,
     s.txn_time,
     d.status
FROM `stream` AS `s` INNER JOIN `dim` FOR SYSTEM_TIME AS OF `s`.`txn_time` AS `d`
ON
     `s`.`id` = `d`.`id`
WHERE
     `d`.`status` = 'OK' 
UNION ALL
SELECT
     s.id,
     s.name,
     s.txn_time,
     d.status
FROM `stream` AS `s` INNER JOIN `dim` FOR SYSTEM_TIME AS OF `s`.`txn_time` AS `d`
ON
     `s`.`id` = `d`.`id`
WHERE
     `d`.`status` = 'NOT_EXISTS';

In this situation the planner will pushdown the filter condition into the table part of the lookup join, but the structure of the lookup joins stays the same, e.g, they will have same digests with different table / temporal table.

This is the problem since the Calcite Volcano optimizer will register them equivalent because it does so using the digest of the relation nodes.

This introduces the bug because when optimizing the Union, both parts of the query will be treated same (even though we have different where clauses) and the found cheapest plan will be same for both lookup joins.

You can also see the effect if you put non-existing filter first the result will be empty, because the first lookupjoin is also used for the second part of the union.

Alternative Solutions

The better solution would be to improve the LookupJoin expression to also include the filter condition into the table name. For example,

LookupJoin(table=[default_catalog.default_database.dim, filter=[<PUSHED-DOWN-FILTERS>]], joinType=[InnerJoin], lookup=[id=id], select=[id, name, txn_time, id], upsertKey=[[0]])

But this would require many refactoring, and mainly in the tests.

In this PR, I have opted for adding another filter with pushed down filter conditions if the LookupJoin contains a table with filter pushdowns.

Brief change log

• Add unit test to reproduce the test
• Add (one alternative) fix to resolve the bug

Verifying this change

The change adds test case that reproduces the bug that can be verified by the fixes.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): no
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? not applicable

[flinkbot]

CI report:

• 11a51dc Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[davidradl]

nit: the Jira number has an extra 0 at the end in the PR title

[davidradl]

nit: normal practise is to list all of the import packages.

[morazow]

Thanks @davidradl, addressed 👍

[davidradl]

@morazow If I am understanding this correctly, the filtered pushed down joins cause the Volcano planner to uniquely identify
each look up join.

I wonder what happens if one of the lookup joins supports filter push down but the other doesn't? Does the include the name of the source as well as the filter? It would be good to have tests for when both sources support filter pushdown, neither do and only 1 does.

[morazow]

Thanks @davidradl, addressed your findings, please have another look 🤝

the filtered pushed down joins cause the Volcano planner to uniquely identify each look up join.

Yes, but based on the getDigests method, which results in a string like below:

LookupJoin(table=[default_catalog.default_database.dim], joinType=[InnerJoin], lookup=[id=id], select=[id, name, txn_time, id], upsertKey=[[0]])

Here we have the table name, but no information on the pushed down filter conditions. For this issue to happen, the lookup table's should be same (since it is in the digest), so we cannot test if one part supports and other part doesn't test case.

Additionally, if the source doesn't support the filter pushdown it will be represented in the digest with the where keyword. This will not also reproduce the bug because the digests are different.

[xuyangzhong]

Thanks for digging into this bug and driving this fix. I just left some comments.

[xuyangzhong]

I'm wondering should PartitionPushDownSpec also need to be added into this part.

[morazow]

Added 👍

[xuyangzhong]

nit what about filterPushedDown?

[xuyangzhong]

Could you please move this test to LookupJoinITCase?

[morazow]

Hello @xuyangzhong ,

Thanks for the review! I have addressed your suggestions, please have a look. I put them to separate commit for now, later I'll squash it 👍

[xuyangzhong]

Thanks for the update. Please forgive me for missing your message... I have one comment as well.

[xuyangzhong]

What about reusing this item table with:

val tableDesc: String = temporalTable match {
  case t: TableSourceTable => t.getQualifiedName.asScala.mkString(", ")
  case t: LegacyTableSourceTable[_] => t.tableIdentifier.asSummaryString()
}

super
  .explainTerms(pw)
  .item("table", tableDesc)
  ...

Although the plans for some tests may change, this will not actually affect compatibility, and it can be aligned with the digest in CommonPhysicalTableSourceScan.

[github-actions]

This PR is being marked as stale since it has not had any activity in the last 90 days.
If you would like to keep this PR alive, please leave a comment asking for a review.
If the PR has merge conflicts, update it with the latest from the base branch.

If you are having difficulty finding a reviewer, please reach out to the
community, contact details can be found here: https://flink.apache.org/what-is-flink/community/

If this PR is no longer valid or desired, please feel free to close it.
If no activity occurs in the next 30 days, it will be automatically closed.

[github-actions]

This PR has been closed since it has not had any activity in 120 days.
If you feel like this was a mistake, or you would like to continue working on it,
please feel free to re-open the PR and ask for a review.

[MartijnVisser]

@morazow I picked this up since the bot closed it rather than anyone rejecting it. You're a co-author on the commit, the reproduction and the test case are yours. Happy to hand it back if you'd rather carry it yourself. New PR: #29200

@xuyangzhong on your two comments:

What about reusing this item table with [...] it can be aligned with the digest in CommonPhysicalTableSourceScan

Went this way, but kept the dotted table name. The lookup join now appends TableSourceTable#getSpecDigests to the table item, so it covers every spec instead of filter and partition only, and matches what the scan digests. getQualifiedName.mkString(", ") would have changed all 138 lookup join plan lines in the goldens instead of 56.

Could you please move this test to LookupJoinITCase?

It ended up as a TableTestProgram in LookupJoinSemanticTests instead, since the planner AGENTS.md prefers semantic tests over ITCase. Both branch orderings are covered, which matters because with the non-matching filter first the query silently returns nothing.

One thing the new PR does not fix: the scan also digests its table hints, so two lookup joins differing only in an OPTIONS hint still get merged. Same class of bug, separate discriminator, I'll file it separately.

[morazow]

Amazing! Thanks @MartijnVisser and @xuyangzhong for reviews 🤝