[FLINK-28568][table-runtime] Implements a new lookup join operator (sync mode only) with state to eliminate the non determinism #20324

lincoln-lil · 2022-07-21T07:36:27Z

What is the purpose of the change

This is a followup implementation of FLINK-28570 which introduces a new lookup join operator (sync mode only) with state to eliminate the non determinism.

Brief change log

add KeyedLookupJoinWrapper to process I/+U vs D/-U differently
add ListenableCollector to offer a callback when original records were collected

Verifying this change

newly added KeyedLookupJoinHarnessTest and existing LookupJoinITCase、AsyncLookupJoinITCase

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)

flinkbot · 2022-07-21T07:50:45Z

CI report:

50f6719 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

lincoln-lil · 2022-08-06T13:11:34Z

rebased latest master branch

godfreyhe

Thanks for the contribution, I left some comments

godfreyhe · 2022-08-07T02:47:13Z

...able-runtime/src/main/java/org/apache/flink/table/runtime/collector/ListenableCollector.java

+ */
+@Internal
+public abstract class ListenableCollector<T> extends TableFunctionCollector<T> {
+    private CollectListener<T> collectListener;


we should mark it as @Nullable

godfreyhe · 2022-08-07T02:48:44Z

...ntime/src/test/java/org/apache/flink/table/runtime/operators/join/LookupJoinHarnessTest.java

     * TableFunctionCollector} which combines left and right into a JoinedRowData.
     */
-    public static final class TestingFetcherCollector extends TableFunctionCollector {
+    public static final class TestingFetcherCollector extends ListenableCollector {


The type should be ListenableCollector<RowData> ?

godfreyhe · 2022-08-07T02:53:49Z

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

+
+        List<Integer> refKeys =
+                allLookupKeys.entrySet().stream()
+                        .filter(


directly filter the FieldRefLookupKeys using filter(key -> (key.getValue() instanceof LookupJoinUtil.FieldRefLookupKey)) ?

godfreyhe · 2022-08-07T02:56:54Z

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

+                new KeyedProcessOperator<>(keyedLookupJoinWrapper);
+
+        List<Integer> refKeys =
+                allLookupKeys.entrySet().stream()


use allLookupKeys.values() can avoid key.getValue()

godfreyhe · 2022-08-07T03:03:12Z

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

+        OneInputTransformation<RowData, RowData> transform =
+                ExecNodeUtil.createOneInputTransformation(
+                        partitionedTransform,
+                        createTransformationMeta(LOOKUP_JOIN_TRANSFORMATION, config),


we should define another operatorName for the Transformation, Because LOOKUP_JOIN_TRANSFORMATION has used for the join Transformation

godfreyhe · 2022-08-07T03:16:22Z

...src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/AsyncLookupJoinITCase.scala

  }

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


Do we have any IT case to verify the change? this pr aims to support sync LookupJoin

this case can cover the change, the legacy source can provide both sync and async functions, so it can fallback to sync lookup function with state.

I've added more cases in LookupJoinITCase

lincoln-lil

@godfreyhe thank you for reviewing this! I've address your comments and updated the pr

lincoln-lil · 2022-08-07T08:33:48Z

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

+
+        List<Integer> refKeys =
+                allLookupKeys.entrySet().stream()
+                        .filter(


lincoln-lil · 2022-08-07T08:34:00Z

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

+        OneInputTransformation<RowData, RowData> transform =
+                ExecNodeUtil.createOneInputTransformation(
+                        partitionedTransform,
+                        createTransformationMeta(LOOKUP_JOIN_TRANSFORMATION, config),


lincoln-lil · 2022-08-07T08:36:29Z

...src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/AsyncLookupJoinITCase.scala

  }

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


this case can cover the change, the legacy source can provide both sync and async functions, so it can fallback to sync lookup function with state.

godfreyhe · 2022-08-08T02:21:35Z

.../test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest.java

+    }
+
+    @Test
+    public void testAggAndAllConstantLookupKeyWithTryResolveMode() {


This method should be moved to LookupJoinTest

lincoln-lil

@godfreyhe I've rebased master to resolve conflicts and added more cases according to your comments, thanks again !

lincoln-lil · 2022-08-08T02:35:06Z

.../test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinJsonPlanTest.java

+    }
+
+    @Test
+    public void testAggAndAllConstantLookupKeyWithTryResolveMode() {


lincoln-lil · 2022-08-08T02:50:01Z

...src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/AsyncLookupJoinITCase.scala

  }

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


I've added more cases in LookupJoinITCase

godfreyhe

Thanks for the update, I left some comments

godfreyhe · 2022-08-08T06:25:46Z

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

+        return transform;
    }

    private LogicalType getLookupKeyLogicalType(


this method can be removed

Yes, this newly method from master is unused, I'll remove it

godfreyhe · 2022-08-08T06:27:46Z

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


    public static final String LOOKUP_JOIN_TRANSFORMATION = "lookup-join";

+    public static final String LOOKUP_JOIN_WITH_STATE_TRANSFORMATION = "lookup-join-with-state";


how about lookup-join-with-materialize ?

changed to 'lookup-join-materialize', omit the word 'with'

godfreyhe · 2022-08-08T06:31:35Z

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

-                        ? (RowType) toLogicalType(temporalTableOutputType.get())
-                        : tableSourceRowType;
-        GeneratedCollector<TableFunctionCollector<RowData>> generatedCollector =
+                getProjectionRowTypeOnTemporalTable(relBuilder);


we can call the getProjectionRowTypeOnTemporalTable method in if (projectionOnTemporalTable != null) branch, and the implementation of getProjectionRowTypeOnTemporalTable method can be simplified

There is some duplicated code in createAsyncLookupJoin method

godfreyhe · 2022-08-08T06:33:33Z

...able-runtime/src/main/java/org/apache/flink/table/runtime/collector/ListenableCollector.java

+public abstract class ListenableCollector<T> extends TableFunctionCollector<T> {
+    @Nullable private CollectListener<T> collectListener;
+
+    public void setCollectListener(CollectListener<T> collectListener) {


nit: @nullable CollectListener collectListener

godfreyhe · 2022-08-08T06:33:59Z

...ntime/src/test/java/org/apache/flink/table/runtime/operators/join/LookupJoinHarnessTest.java

-            RowData right = (RowData) record;
+            RowData right = record;
+            getCollectListener()
+                    .ifPresent(listener -> ((CollectListener) listener).onCollect(record));


cast is no needed here

godfreyhe

Thanks for the update, LGTM. Please rebase master, the failure test had been fixed

…ync mode only) with state to eliminate non-deterministic result

…no for ListenableCollector

lincoln-lil · 2022-08-08T12:08:14Z

rebased latest master to get rid of failed yarn it case

lincoln-lil · 2022-08-09T01:25:49Z

@lincoln-lil run azure

lincoln-lil · 2022-08-09T02:08:20Z

An irrelevant failure case of es sink https://issues.apache.org/jira/browse/FLINK-28877

godfreyhe · 2022-08-09T03:18:37Z

An irrelevant failure case of es sink https://issues.apache.org/jira/browse/FLINK-28877

I will merge it

…ync mode only) with state to eliminate non-deterministic result This closes apache#20324

flinkbot added the component=TableSQL/Runtime label Jul 21, 2022

lincoln-lil force-pushed the FLINK-28568 branch 3 times, most recently from 7fa8139 to 236db88 Compare August 6, 2022 13:10

godfreyhe reviewed Aug 7, 2022

View reviewed changes

lincoln-lil force-pushed the FLINK-28568 branch from 236db88 to 2232c0d Compare August 7, 2022 08:42

lincoln-lil commented Aug 7, 2022

View reviewed changes

lincoln-lil force-pushed the FLINK-28568 branch from f8cf4e2 to 3329be5 Compare August 7, 2022 15:13

godfreyhe reviewed Aug 8, 2022

View reviewed changes

lincoln-lil force-pushed the FLINK-28568 branch from 542a420 to e257c65 Compare August 8, 2022 03:07

lincoln-lil commented Aug 8, 2022

View reviewed changes

godfreyhe reviewed Aug 8, 2022

View reviewed changes

lincoln-lil force-pushed the FLINK-28568 branch from fd845d9 to c502d15 Compare August 8, 2022 06:53

godfreyhe approved these changes Aug 8, 2022

View reviewed changes

lincoln-lil added 10 commits August 8, 2022 20:05

[FLINK-28568][table-runtime] Implements a new lookup join operator (s…

331f99c

…ync mode only) with state to eliminate non-deterministic result

address comments and rebase master

3e67f54

add nullable annotation

99c0915

fix fail test due to incorrectly code simplification and add more cases

a1f61d6

address comments and fix the case that lookup key contains constants

28a6351

rebase master to resolve conflicts and add more case

0751fd2

fix merge error after rebase

21cd970

address comments

227d2c6

remove unnecessary cast for LookupJoinHarnessTest and add nullable an…

4defa85

…no for ListenableCollector

minor refactor on right output row type info

50f6719

lincoln-lil force-pushed the FLINK-28568 branch from 2d453dc to 50f6719 Compare August 8, 2022 12:06

godfreyhe closed this in 3f18caf Aug 9, 2022

huangxiaofeng10047 pushed a commit to huangxiaofeng10047/flink that referenced this pull request Nov 3, 2022

[FLINK-28568][table-runtime] Implements a new lookup join operator (s…

22d32a1

…ync mode only) with state to eliminate non-deterministic result This closes apache#20324


		public static final String LOOKUP_JOIN_TRANSFORMATION = "lookup-join";

		public static final String LOOKUP_JOIN_WITH_STATE_TRANSFORMATION = "lookup-join-with-state";

[FLINK-28568][table-runtime] Implements a new lookup join operator (sync mode only) with state to eliminate the non determinism #20324

[FLINK-28568][table-runtime] Implements a new lookup join operator (sync mode only) with state to eliminate the non determinism #20324

Uh oh!

Conversation

lincoln-lil commented Jul 21, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

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

Uh oh!

flinkbot commented Jul 21, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

lincoln-lil commented Aug 6, 2022

Uh oh!

godfreyhe left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lincoln-lil left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lincoln-lil left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

godfreyhe left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lincoln-lil commented Jul 21, 2022 •

edited

Loading

flinkbot commented Jul 21, 2022 •

edited

Loading