[FLINK-28848][table-planner] Introduces LOOKUP join hint to support delayed retry for lookup join (table alias unsupported in hint) #20482

lincoln-lil · 2022-08-07T09:59:27Z

What is the purpose of the change

Introduces LOOKUP join hint to support delayed retry for lookup join (table alias unsupported in hint)
This is the main part of FLINK-28779 to implement FLIP-234: Support Retryable Lookup Join To Solve Delayed Updates Issue In External Systems.
The join hint based on [FLINK-28682][table-planner] support join hint in batch rules #20359, and adds a new hint 'LOOKUP' described in FLIP-234.

Brief change log

Add new hint LOOKUP
Add related option class and validation
Add related tests for new classes
Adapt to async operator with retry and also adds a new RetryableLookupFunctionDelegator to wrap user's sync lookup func to support retry

Verifying this change

LookupJoinHintSpecSerdeTest
LookupJoinHintSpecTest
LookupJoinJsonPlanTest
LookupJoinTest
RetryableLookupFunctionDelegatorTest
more ITCases will be updated in Add it cases with lookup hint using new lookup function api for lookup join

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? (yes)
If yes, how is the feature documented? (docs)

flinkbot · 2022-08-07T10:10:08Z

CI report:

e64a0ed Azure: CANCELED
adb0949 Azure: PENDING

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

lincoln-lil · 2022-08-08T04:08:08Z

rebased master and updated the pr

lincoln-lil · 2022-08-08T04:30:54Z

...nk-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LookupJoinUtil.java

-                            partialCachingLookupProvider.getCache(),
-                            partialCachingLookupProvider.createAsyncLookupFunction());
+                    asyncLookupFunctions =
+                            new CachingAsyncLookupFunction(


this function should be refactored to create nested delegators of user lookup func

[CachingLookupFunction]
|
[RetryableLookupFunctionDelegator]
|
userLookupFunc

I'll update this

godfreyhe

Thanks for the contribution @lincoln-lil , I left some comments, and could you add some hint propagation tests ?

godfreyhe · 2022-08-08T09:38:12Z

...nk-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LookupJoinUtil.java

+            LookupJoinHintSpec joinHintSpec,
+            boolean upsertMaterialize) {
+        // async & sync lookup candidates
+        Tuple2<UserDefinedFunction, UserDefinedFunction> lookupFunctions = new Tuple2<>();


introduce a new class to represent the functions, which is more clear and readable

ok, will add a private static class LookupFunctionCandidates

godfreyhe · 2022-08-08T09:41:14Z

...nner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.scala

@@ -543,6 +554,201 @@ class LookupJoinTest(legacyTableSource: Boolean) extends TableTestBase with Seri
    util.verifyExecPlan(sql)
  }

+  @Test
+  def testInvalidJoinHint(): Unit = {


please add some tests which hints refers an inner table name

godfreyhe · 2022-08-08T09:42:19Z

...rc/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecLookupJoin.java

@@ -179,6 +181,9 @@ public abstract class CommonExecLookupJoin extends ExecNodeBase<RowData>
    @JsonProperty(FIELD_NAME_INPUT_CHANGELOG_MODE)
    private final ChangelogMode inputChangelogMode;

+    @JsonProperty(FIELD_NAME_JOIN_HINT)
+    private final @Nullable LookupJoinHintSpec joinHintSpec;


ignore if joinHintSpec is null

godfreyhe · 2022-08-08T09:47:45Z

...nner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.scala

+        |JOIN LookupTable FOR SYSTEM_TIME AS OF T.proctime AS D
+        | ON T.a = D.id
+        |""".stripMargin,
+      "Invalid LOOKUP hint options, parsing error: Could not parse value 'yes' for key 'async'",


parsing error will cause misunderstanding, just remove it

will change to "Invalid LOOKUP hint options: Could not..."

godfreyhe · 2022-08-08T09:56:02Z

...er/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/spec/LookupJoinHintSpec.java

+    private final Long asyncTimeout;
+
+    @JsonProperty(FIELD_NAME_RETRY_PREDICATE)
+    private final @Nullable String retryPredicate;


ignore empty: @JsonInclude(JsonInclude.Include.NON_NULL)

lincoln-lil

@godfreyhe thanks for reviewing this! I'll update the pr according to your comments

lincoln-lil · 2022-08-08T10:21:05Z

...rc/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecLookupJoin.java

@@ -179,6 +181,9 @@ public abstract class CommonExecLookupJoin extends ExecNodeBase<RowData>
    @JsonProperty(FIELD_NAME_INPUT_CHANGELOG_MODE)
    private final ChangelogMode inputChangelogMode;

+    @JsonProperty(FIELD_NAME_JOIN_HINT)
+    private final @Nullable LookupJoinHintSpec joinHintSpec;


lincoln-lil · 2022-08-08T10:24:45Z

...nk-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LookupJoinUtil.java

+            LookupJoinHintSpec joinHintSpec,
+            boolean upsertMaterialize) {
+        // async & sync lookup candidates
+        Tuple2<UserDefinedFunction, UserDefinedFunction> lookupFunctions = new Tuple2<>();


ok, will add a private static class LookupFunctionCandidates

lincoln-lil · 2022-08-08T10:29:04Z

...nner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.scala

@@ -543,6 +554,201 @@ class LookupJoinTest(legacyTableSource: Boolean) extends TableTestBase with Seri
    util.verifyExecPlan(sql)
  }

+  @Test
+  def testInvalidJoinHint(): Unit = {


lincoln-lil · 2022-08-08T10:29:32Z

...nner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.scala

+        |JOIN LookupTable FOR SYSTEM_TIME AS OF T.proctime AS D
+        | ON T.a = D.id
+        |""".stripMargin,
+      "Invalid LOOKUP hint options, parsing error: Could not parse value 'yes' for key 'async'",


will change to "Invalid LOOKUP hint options: Could not..."

godfreyhe

LGTM

…elayed retry for lookup join (table alias unsupported in hint) This is the main part of FLINK-28779 to implement FLIP-234: Support Retryable Lookup Join To Solve Delayed Updates Issue In External Systems

…on type instead of one-level parent class compartion in LookupJoinCodeGenerator This bug can be reproduced by AsyncLookupJoinITCase#testAsyncJoinTemporalTableWithLookupThresholdWithInsufficientRetry when caching is disabled in FLINK-28849

…retry timer to prevent unexpected incomplete element during finish phase in AsyncWaitOperator It is hard to reproduce this in runtime tests, but occasionally happens in AsyncLookupJoinITCase#testAsyncJoinTemporalTableWithLookupThresholdWithSufficientRetry of FLINK-28849. It's better to add a separte test in runtime.

…okup and add more tests

lincoln-lil · 2022-08-10T01:59:56Z

@flinkbot run azure

lincoln-lil · 2022-08-10T03:04:10Z

@flinkbot run azure

lincoln-lil · 2022-08-10T04:54:29Z

there're several merge conflicts for the latest master, so need to verify the merged pr completely, the local verify is running now

lincoln-lil · 2022-08-10T05:14:48Z

local verify ok for table modules(the ci pipeline was too slow)

…on type instead of one-level parent class compartion in LookupJoinCodeGenerator This bug can be reproduced by AsyncLookupJoinITCase#testAsyncJoinTemporalTableWithLookupThresholdWithInsufficientRetry when caching is disabled in FLINK-28849 This closes #20482

…retry timer to prevent unexpected incomplete element during finish phase in AsyncWaitOperator It is hard to reproduce this in runtime tests, but occasionally happens in AsyncLookupJoinITCase#testAsyncJoinTemporalTableWithLookupThresholdWithSufficientRetry of FLINK-28849. It's better to add a separate test in runtime. This closes #20482

…okup and add more tests Disable retry on async because of two problems need to be resolved first This closes #20482

…elayed retry for lookup join (table alias unsupported in hint) This is the main part of FLINK-28779 to implement FLIP-234: Support Retryable Lookup Join To Solve Delayed Updates Issue In External Systems This closes apache#20482

…on type instead of one-level parent class compartion in LookupJoinCodeGenerator This bug can be reproduced by AsyncLookupJoinITCase#testAsyncJoinTemporalTableWithLookupThresholdWithInsufficientRetry when caching is disabled in FLINK-28849 This closes apache#20482

…retry timer to prevent unexpected incomplete element during finish phase in AsyncWaitOperator It is hard to reproduce this in runtime tests, but occasionally happens in AsyncLookupJoinITCase#testAsyncJoinTemporalTableWithLookupThresholdWithSufficientRetry of FLINK-28849. It's better to add a separate test in runtime. This closes apache#20482

…okup and add more tests Disable retry on async because of two problems need to be resolved first This closes apache#20482

flinkbot added the component=<none> label Aug 7, 2022

lincoln-lil force-pushed the FLINK-28848 branch from 6526216 to ac69d96 Compare August 8, 2022 04:07

lincoln-lil commented Aug 8, 2022

View reviewed changes

godfreyhe reviewed Aug 8, 2022

View reviewed changes

lincoln-lil commented Aug 8, 2022

View reviewed changes

lincoln-lil force-pushed the FLINK-28848 branch 3 times, most recently from 09cbf8a to 06da0fb Compare August 9, 2022 08:17

godfreyhe approved these changes Aug 9, 2022

View reviewed changes

lincoln-lil added 5 commits August 10, 2022 09:40

[FLINK-28848][table-planner] Introduces LOOKUP join hint to support d…

6244d20

…elayed retry for lookup join (table alias unsupported in hint) This is the main part of FLINK-28779 to implement FLIP-234: Support Retryable Lookup Join To Solve Delayed Updates Issue In External Systems

[FLINK-28849][table-planner] Fix errors when enable retry on async lo…

4a44ef5

…okup and add more tests

disable retry on async because of two problems need to be resolved first

b6daeb6

lincoln-lil force-pushed the FLINK-28848 branch from 6f7e54f to b6daeb6 Compare August 10, 2022 01:52

fix tests after merge

575d431

lincoln-lil added 2 commits August 10, 2022 12:13

update test xml for LookupJoinTest

e64a0ed

fix LookupJoinITCase due to merge error: lost new test lookup funcs

adb0949

godfreyhe closed this in 8b25b96 Aug 10, 2022

godfreyhe pushed a commit that referenced this pull request Aug 10, 2022

[FLINK-28849][table-planner] Fix errors when enable retry on async lo…

c5b5d43

…okup and add more tests Disable retry on async because of two problems need to be resolved first This closes #20482

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-28848][table-planner] Introduces LOOKUP join hint to support delayed retry for lookup join (table alias unsupported in hint) #20482

[FLINK-28848][table-planner] Introduces LOOKUP join hint to support delayed retry for lookup join (table alias unsupported in hint) #20482

lincoln-lil commented Aug 7, 2022

flinkbot commented Aug 7, 2022 •

edited

lincoln-lil commented Aug 8, 2022

lincoln-lil Aug 8, 2022

godfreyhe left a comment

godfreyhe Aug 8, 2022

lincoln-lil Aug 8, 2022

godfreyhe Aug 8, 2022

lincoln-lil Aug 8, 2022

godfreyhe Aug 8, 2022

lincoln-lil Aug 8, 2022

godfreyhe Aug 8, 2022

lincoln-lil Aug 8, 2022

godfreyhe Aug 8, 2022

lincoln-lil left a comment

lincoln-lil Aug 8, 2022

lincoln-lil Aug 8, 2022

lincoln-lil Aug 8, 2022

lincoln-lil Aug 8, 2022

godfreyhe left a comment

lincoln-lil commented Aug 10, 2022

lincoln-lil commented Aug 10, 2022

lincoln-lil commented Aug 10, 2022 •

edited

lincoln-lil commented Aug 10, 2022

[FLINK-28848][table-planner] Introduces LOOKUP join hint to support delayed retry for lookup join (table alias unsupported in hint) #20482

[FLINK-28848][table-planner] Introduces LOOKUP join hint to support delayed retry for lookup join (table alias unsupported in hint) #20482

Conversation

lincoln-lil commented Aug 7, 2022

What is the purpose of the change

Brief change log

Verifying this change

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

Documentation

flinkbot commented Aug 7, 2022 • edited

CI report:

lincoln-lil commented Aug 8, 2022

Choose a reason for hiding this comment

godfreyhe left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lincoln-lil left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

godfreyhe left a comment

Choose a reason for hiding this comment

lincoln-lil commented Aug 10, 2022

lincoln-lil commented Aug 10, 2022

lincoln-lil commented Aug 10, 2022 • edited

lincoln-lil commented Aug 10, 2022

flinkbot commented Aug 7, 2022 •

edited

lincoln-lil commented Aug 10, 2022 •

edited