[FLINK-36808][table-planner] Include pushed-down source abilities in the lookup join digest - #29200
MartijnVisser wants to merge 1 commit into
Conversation
e9a490b to
684ea1b
Compare
| import java.util.List; | ||
|
|
||
| /** Semantic tests for {@link StreamExecLookupJoin}. */ | ||
| public class LookupJoinSemanticTests extends SemanticTestBase { |
There was a problem hiding this comment.
there is JoinSemanticTests
not sure if we need a dedicated class for LookupJoin
There was a problem hiding this comment.
JoinSemanticTests is for StreamExecJoin only. Its eight programs are all plain stream-stream joins, there is no SYSTEM_TIME or LOOKUP in any of them. Lateral snapshot join and multi join are also joins and both got their own class rather than going in there, so I kept this one.
You did make me look at where the programs live though. They were in LookupJoinTestPrograms, which is the restore-test file, and the two sources I added only exist because a semantic test cannot take restore data. Moved them to LookupJoinSemanticTestPrograms, same split FLINK-38720 did for joins.
| """ | ||
| |SELECT s.a, s.b, s.proctime, d.status | ||
| |FROM MyTable AS s | ||
| |JOIN LookupTableWithFilterableFields FOR SYSTEM_TIME AS OF s.proctime AS d | ||
| |ON s.a = d.id | ||
| |WHERE d.status = 'OK' | ||
| |UNION ALL | ||
| |SELECT s.a, s.b, s.proctime, d.status | ||
| |FROM MyTable AS s | ||
| |JOIN LookupTableWithFilterableFields FOR SYSTEM_TIME AS OF s.proctime AS d | ||
| |ON s.a = d.id | ||
| |WHERE d.status = 'KO' | ||
| """.stripMargin | ||
|
|
||
| util.verifyExecPlan(sql) | ||
| } | ||
|
|
||
| @Test | ||
| def testJoinFilterableTemporalTableWithUnionSameFilter(): Unit = { | ||
| // Counterpart to testJoinFilterableTemporalTableWithUnion: with the same filter on both sides | ||
| // the two lookup joins really are equivalent and must still be reused. Guards against a fix | ||
| // that simply makes every lookup join digest unique. | ||
| val sql = | ||
| """ | ||
| |SELECT s.a, s.b, s.proctime, d.status | ||
| |FROM MyTable AS s | ||
| |JOIN LookupTableWithFilterableFields FOR SYSTEM_TIME AS OF s.proctime AS d |
There was a problem hiding this comment.
do these test trigger something that semantic can not?
Or why do we need them here?
There was a problem hiding this comment.
The same-filter one cannot be a semantic test. Reuse or not, the UNION ALL still has two inputs, so the sink sees each row twice either way and the output is identical. Only the plan shows whether the two joins collapsed. It guards the other direction of this fix, a digest that is too unique and silently kills legitimate subplan reuse.
ScanReuseTest has the same pair on the scan side, testProjectWithFilterPushDown and testProjectReuseWithFilterPushDown, both plan tests.
…the lookup join digest A lookup join rendered only the temporal table's ObjectIdentifier into its digest, while a TableSourceScan digests RelOptTable#getQualifiedName, which TableSourceTable extends with the digest of every pushed-down SourceAbilitySpec. Two lookup joins on the same table with different filters pushed into them therefore had identical digests and were merged, so one branch of a UNION ALL emitted the other branch's rows, or none at all. Append the ability spec digests to the table item so the pushed-down filter is part of the node's identity. Empty digests are left out: they carry no information and would otherwise add noise to every lookup join plan. This does not close the gap with the scan entirely: the scan also digests its table hints, so two lookup joins differing only in an OPTIONS hint still collapse. That is a separate discriminator and is left for a follow-up. The diagnosis and the reproduction are Muhammet Orazov's, from apache#26514, which the stale bot closed rather than rejected; the test case here is his. Putting the spec digests in the existing table item instead of adding a separate one is what Xuyang Zhong asked for in review on that PR. Co-authored-by: Muhammet Orazov <m.orazow@gmail.com> Generated-by: Claude Code (Claude Opus 5)
684ea1b to
1adbb80
Compare
3 participants
What is the purpose of the change
A lookup join put only the temporal table's
ObjectIdentifierin its digest, while aTableSourceScandigestsRelOptTable#getQualifiedName, whichTableSourceTableextends with the digest of every pushed-downSourceAbilitySpec. Once a dim-side filter is pushed into the source there is nothing left to tell two lookup joins apart, so the planner merges them. AUNION ALLover the same table with two different filters then emits one branch's rows under the other branch's filter, or nothing at all when the non-matching branch comes first.This appends the ability spec digests to the
tableitem. It does not close the gap with the scan entirely: the scan also digests its table hints, so two lookup joins differing only in anOPTIONShint still merge. That is a separate discriminator and I'll file a follow-up.The diagnosis and the reproduction are @morazow's, from #26514, which the stale bot closed rather
than rejected, and the test case here is his, so he is a co-author on the commit. Putting the spec
digests in the existing
tableitem instead of adding a separate one is what @xuyangzhong askedfor in review on that PR. Happy to hand this back to @morazow if he'd rather carry it.
Brief change log
CommonPhysicalLookupJoin.explainTermsappendsTableSourceTable#getSpecDigeststo thetableitem, which covers stream and batch through the shared base classLookupJoinSemanticTestscovers both branch orderings, since the wrong order silently returns nothingdeterminism.mdpages updated for the richertableitemVerifying this change
This change added tests and can be verified as follows:
LookupJoinTest.testJoinFilterableTemporalTableWithUnionexpects twoLookupJoinnodes; before the fix the plan collapses to one plusReused(reference_id=[1])LookupJoinTest.testJoinFilterableTemporalTableWithUnionSameFilterexpects that reuse to survive, so a fix that merely made every digest unique would failLookupJoinSemanticTestsreturns 8 rows instead of 4 before the fix with the matching filter first, and an empty result with it second./mvnw -pl flink-table/flink-table-planner testpasses apart from FLINK-40568, an open flake unrelated to lookup joinsTwo notes for reviewers. The
tableitem now lists the pushed-down abilities exactly asTableSourceScanalready does, which is why plans unrelated to this bug change: projection push-down through a snapshot is routine, and aFilterPushDownSpecis attached whenever a dim-side predicate is convertible even when the source accepts none of it, rendering asfilter=[]. And a plan compiled before this fix has the merge baked in, so affected users have to recompile.Does this pull request potentially affect one of the following parts:
@Public(Evolving): noDocumentation
Was generative AI tooling used to co-author this PR?
Generated-by: Claude Code (Claude Opus 5)