[FLINK-25990][table] Expose uid generator for DataStream/Transformation providers #18667

slinkydeveloper · 2022-02-08T16:35:20Z

What is the purpose of the change

Expose uid generator for DataStream/Transformation providers. Also update collect, print, kafka, hive and files connectors to properly set up uids for generated DataStreams/Transformations

flinkbot · 2022-02-08T16:38:06Z

CI report:

ca7a596 Azure: FAILURE
e57a8f4 Azure: PENDING

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

flinkbot · 2022-02-08T16:39:17Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit 8926e35 (Tue Feb 08 16:39:17 UTC 2022)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

slinkydeveloper · 2022-02-09T11:02:13Z

@flinkbot run azure

twalthr

Thanks for the PR @slinkydeveloper. So far we have not added a single test to the uid business. I'm wondering whether it makes sense to have a temporary solution by adapting org.apache.flink.table.planner.utils.JsonPlanTestBase#compileSqlAndExecutePlan and traverse the transformations tree to check if all transformations have a uid assigned. What do you think?

...nk-table-api-java-bridge/src/main/java/org/apache/flink/table/connector/ProviderContext.java

...nner/src/main/java/org/apache/flink/table/planner/connectors/TransformationScanProvider.java

...nner/src/main/java/org/apache/flink/table/planner/connectors/TransformationSinkProvider.java

twalthr · 2022-02-10T13:18:26Z

...nner/src/main/java/org/apache/flink/table/planner/connectors/TransformationSinkProvider.java

    Transformation<?> createTransformation(Context context);

    /** Context for {@link #createTransformation(Context)}. */
-    interface Context {
+    interface Context extends ProviderContext {


Rename to TransformationProviderContext?

isn't this already implied by the fact that this context is within TransformationSinkProvider?

true, but if you implement a connector, you have a couple of Context classes and the code could become quite messy.

But then the solution is just to don't have static import for Context?

twalthr · 2022-02-10T13:19:50Z

...able-planner/src/main/java/org/apache/flink/table/planner/connectors/CollectDynamicSink.java

@@ -93,39 +98,40 @@ public ChangelogMode getChangelogMode(ChangelogMode requestedMode) {

    @Override
    public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
-        return (DataStreamSinkProvider)
-                inputStream -> {


did we break the API for lambdas now?

yes 😢 the only way to not break it is to not default implement the new method

...e-planner/src/main/java/org/apache/flink/table/planner/connectors/ExternalDynamicSource.java

...e-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java

…xt for generating uid and use it in DataStreamScanProvider, DataStreamSinkProvider, TransformationScanProvider, TransformationSinkProvider Signed-off-by: slinkydeveloper <francescoguard@gmail.com>

Signed-off-by: slinkydeveloper <francescoguard@gmail.com>

…ormation uids Signed-off-by: slinkydeveloper <francescoguard@gmail.com>

twalthr

Thanks for the update @slinkydeveloper. I added the last comments. Should be good in the next iteration.

twalthr · 2022-02-11T08:57:49Z

...java/org/apache/flink/table/planner/runtime/stream/jsonplan/DeduplicationJsonPlanITCase.java

@@ -71,6 +71,7 @@ public void testDeduplication() throws Exception {
        tableEnv.getConfig()
                .getConfiguration()
                .setInteger(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, 1);
+        checkTransformationUids(compiledPlan);


Is there a reason why not to use compileSqlAndExecutePlan? I think the configuration can also be set before without side effects?

I honestly didn't knew if this affects the plan or not, so i defaulted to just keeping the code as it is. I can change it if you want

twalthr · 2022-02-11T09:09:19Z

...lanner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/TransformationsTest.java

-    }
-
-    @Test
-    public void testBatchTransformationScanProvider() {


the tests are important, they test the propagation of boundedness. please readd them

as discussed offline, we can remove them as they're only testing mocks now

twalthr · 2022-02-11T09:12:46Z

...nner/src/main/java/org/apache/flink/table/planner/connectors/TransformationSinkProvider.java

    Transformation<?> createTransformation(Context context);

    /** Context for {@link #createTransformation(Context)}. */
-    interface Context {
+    interface Context extends ProviderContext {


true, but if you implement a connector, you have a couple of Context classes and the code could become quite messy.

Signed-off-by: slinkydeveloper <francescoguard@gmail.com>

twalthr

LGTM

@FunctionalInterface

…ng uids in DataStream/Transformation Scan/SinkProvider This might break existing implementations if the provider was implemented as a Java lambda. Please note that the interface was not annotated as a @FunctionalInterface. This closes apache#18667.

@FunctionalInterface

…ng uids in DataStream/Transformation Scan/SinkProvider This might break existing implementations if the provider was implemented as a Java lambda. Please note that the interface was not annotated as a @FunctionalInterface. This closes apache#18667.

rmetzger added the component=TableSQL/API label Feb 8, 2022

slinkydeveloper force-pushed the FLINK-25990 branch 2 times, most recently from db2efa4 to 4af7f82 Compare February 9, 2022 08:33

twalthr requested changes Feb 10, 2022

View reviewed changes

slinkydeveloper added 6 commits February 10, 2022 15:55

[FLINK-25990][table-api-java-bridge][table-planner] Add ProviderConte…

4e61da1

…xt for generating uid and use it in DataStreamScanProvider, DataStreamSinkProvider, TransformationScanProvider, TransformationSinkProvider Signed-off-by: slinkydeveloper <francescoguard@gmail.com>

[fixup][table-planner] Use the new ProviderContext based method

1f4fbd1

Signed-off-by: slinkydeveloper <francescoguard@gmail.com>

[fixup][connector-kafka] Use the new ProviderContext based method

d4c6180

Signed-off-by: slinkydeveloper <francescoguard@gmail.com>

[fixup][connector-hive] Use the new ProviderContext based method

3405603

Signed-off-by: slinkydeveloper <francescoguard@gmail.com>

[fixup][connector-files] Use the new ProviderContext based method

4a56766

Signed-off-by: slinkydeveloper <francescoguard@gmail.com>

[fixup] Feedback

42c9b0a

Signed-off-by: slinkydeveloper <francescoguard@gmail.com>

slinkydeveloper force-pushed the FLINK-25990 branch from 4af7f82 to f8e3980 Compare February 10, 2022 15:14

[fixup] Now json plan tests are checking that we correctly set transf…

ca7a596

…ormation uids Signed-off-by: slinkydeveloper <francescoguard@gmail.com>

slinkydeveloper force-pushed the FLINK-25990 branch from f8e3980 to ca7a596 Compare February 10, 2022 15:15

twalthr requested changes Feb 11, 2022

View reviewed changes

[fixup] Another small refactor

e57a8f4

Signed-off-by: slinkydeveloper <francescoguard@gmail.com>

twalthr approved these changes Feb 11, 2022

View reviewed changes

twalthr closed this in 3adca15 Feb 11, 2022

slinkydeveloper deleted the FLINK-25990 branch February 11, 2022 15:22

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-25990][table] Expose uid generator for DataStream/Transformation providers #18667

[FLINK-25990][table] Expose uid generator for DataStream/Transformation providers #18667

slinkydeveloper commented Feb 8, 2022

flinkbot commented Feb 8, 2022 •

edited

Loading

flinkbot commented Feb 8, 2022

slinkydeveloper commented Feb 9, 2022

twalthr left a comment

twalthr Feb 10, 2022

slinkydeveloper Feb 10, 2022

twalthr Feb 11, 2022

slinkydeveloper Feb 11, 2022

twalthr Feb 10, 2022

slinkydeveloper Feb 10, 2022

slinkydeveloper Feb 10, 2022

twalthr left a comment

twalthr Feb 11, 2022

slinkydeveloper Feb 11, 2022

twalthr Feb 11, 2022

slinkydeveloper Feb 11, 2022

twalthr Feb 11, 2022

twalthr left a comment

[FLINK-25990][table] Expose uid generator for DataStream/Transformation providers #18667

[FLINK-25990][table] Expose uid generator for DataStream/Transformation providers #18667

Conversation

slinkydeveloper commented Feb 8, 2022

What is the purpose of the change

flinkbot commented Feb 8, 2022 • edited Loading

CI report:

flinkbot commented Feb 8, 2022

Automated Checks

Review Progress

slinkydeveloper commented Feb 9, 2022

twalthr 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

twalthr 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

twalthr left a comment

Choose a reason for hiding this comment

flinkbot commented Feb 8, 2022 •

edited

Loading