[FLINK-24388][table] Modules can provide a table source/sink factory #17384

Airblader · 2021-09-29T12:20:30Z

What is the purpose of the change

This introduces Module#getTableSourceFactory and Module#getTableSinkFactory, which allow providing a corresponding factory when sources/sinks are created. The precedence is as follows:

Factory defined by the catalog
Factory defined by the module
Factory discovery

It may seem more appropriate to give modules the higher precedence, but in practice it's more useful to prefer the specificity of the catalog and have the module act as a fallback (that can override the discovery process fallback).

The changes in FactoryUtil have been kept backwards-compatible, but the old methods have been deprecated.

Brief change log

Provide ModuleManager in FlinkContext
Introduce the new APIs.

Verifying this change

(Please pick either of the following options)

Compatibility is covered through existing tests. A new ModuleITCase has been added to verify the new module APIs.

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): yes
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? JavaDocs

Airblader · 2021-09-29T12:22:21Z

flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/FactoryUtil.java

     */
-    public static DynamicTableSource createTableSource(
-            @Nullable Catalog catalog,
+    public static DynamicTableSource createSource(


I shortened the name here because simply adding an overload (replacing the first argument) would be a breaking change when null is passed (as it would cause an ambiguity).

Can we name it createDynamicTableSource instead? Source sounds like the core Source abstraction.

flinkbot · 2021-09-29T12:23:54Z

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 607bc15 (Wed Sep 29 12:23:54 UTC 2021)

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

flinkbot · 2021-09-29T12:54:51Z

CI report:

f26b757 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build

twalthr

Thanks for the PR @Airblader. I added some comments. Please also update each commit message with a component tag that the commit mostly touches.

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/module/ModuleManager.java

flink-core/src/main/java/org/apache/flink/util/OptionalUtils.java

twalthr · 2021-09-30T07:57:51Z

flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/FactoryUtil.java

     */
-    public static DynamicTableSource createTableSource(
-            @Nullable Catalog catalog,
+    public static DynamicTableSource createSource(


Can we name it createDynamicTableSource instead? Source sounds like the core Source abstraction.

flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/FactoryUtil.java

flink-table/flink-table-common/src/main/java/org/apache/flink/table/module/Module.java

...planner/src/test/java/org/apache/flink/table/planner/runtime/stream/module/ModuleITCase.java

...lanner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecSink.java

...nner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java

twalthr

More comments ;-)

flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/SqlClientTest.java

flink-table/flink-table-common/src/main/java/org/apache/flink/table/module/Module.java

...nner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecSink.java

twalthr

LGTM, I had one last minor thing but can do this while merging once the build is green.

twalthr · 2021-09-30T14:41:08Z

...nner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java

            boolean upsertMaterialize) {
-        final DynamicTableSink tableSink = tableSinkSpec.getTableSink();
+        final DynamicTableSink tableSink = tableSinkSpec.getTableSink(planner);
+        final ChangelogMode inputChangelogMode = tableSink.getChangelogMode(changelogMode);


sorry, but I still fine this confusing. Shouldn't this variable be called changelogMode and the member inputChangelogMode

This closes apache#17384.

Airblader commented Sep 29, 2021

View reviewed changes

Airblader force-pushed the FLINK-24388 branch from 607bc15 to 53268da Compare September 29, 2021 12:30

rmetzger added the component=TableSQL/API label Sep 29, 2021

Airblader force-pushed the FLINK-24388 branch from 53268da to 999215d Compare September 30, 2021 05:47

Airblader marked this pull request as ready for review September 30, 2021 05:49

twalthr reviewed Sep 30, 2021

View reviewed changes

Airblader added 3 commits September 30, 2021 10:45

[hotfix][table-api-java] Mark ModuleManager as internal

0a1cdfa

[hotfix][core] Added OptionalUtils

22fe729

[FLINK-24388][planner] Provide ModuleManager in FlinkContext

ea32c55

Airblader force-pushed the FLINK-24388 branch from 999215d to 7af1ce3 Compare September 30, 2021 10:13

Airblader requested a review from twalthr September 30, 2021 10:13

twalthr reviewed Sep 30, 2021

View reviewed changes

Airblader added 2 commits September 30, 2021 13:46

[FLINK-24388][planner] Introduce Module#getTableSourceFactory

78a92ec

[fixup] Review comments

b26de4e

Airblader force-pushed the FLINK-24388 branch from 7af1ce3 to 1c3a1b0 Compare September 30, 2021 11:52

[FLINK-24388][planner] Introduce Module#getTableSinkFactory

f26b757

Airblader force-pushed the FLINK-24388 branch from 1c3a1b0 to f26b757 Compare September 30, 2021 13:14

Airblader requested a review from twalthr September 30, 2021 13:15

twalthr approved these changes Sep 30, 2021

View reviewed changes

twalthr pushed a commit to twalthr/flink that referenced this pull request Oct 1, 2021

[FLINK-24388][table-planner] Introduce Module#getTableSinkFactory

56de7cd

This closes apache#17384.

twalthr closed this in aa51c8f Oct 1, 2021

Airblader deleted the FLINK-24388 branch October 25, 2021 18:39

niklassemmler pushed a commit to niklassemmler/flink that referenced this pull request Feb 3, 2022

[FLINK-24388][table-planner] Introduce Module#getTableSinkFactory

a4c7369

This closes apache#17384.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[FLINK-24388][table] Modules can provide a table source/sink factory #17384

[FLINK-24388][table] Modules can provide a table source/sink factory #17384

Uh oh!

Airblader commented Sep 29, 2021

Uh oh!

Airblader Sep 29, 2021

Uh oh!

twalthr Sep 30, 2021

Uh oh!

flinkbot commented Sep 29, 2021

Uh oh!

flinkbot commented Sep 29, 2021 •

edited

Loading

Uh oh!

twalthr left a comment

Uh oh!

Uh oh!

Uh oh!

twalthr Sep 30, 2021

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

twalthr left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

twalthr left a comment

Uh oh!

twalthr Sep 30, 2021

Uh oh!

Uh oh!

[FLINK-24388][table] Modules can provide a table source/sink factory #17384

[FLINK-24388][table] Modules can provide a table source/sink factory #17384

Uh oh!

Conversation

Airblader commented Sep 29, 2021

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!

Airblader Sep 29, 2021

Choose a reason for hiding this comment

Uh oh!

twalthr Sep 30, 2021

Choose a reason for hiding this comment

Uh oh!

flinkbot commented Sep 29, 2021

Automated Checks

Review Progress

Uh oh!

flinkbot commented Sep 29, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

twalthr left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

twalthr Sep 30, 2021

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

twalthr left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

twalthr left a comment

Choose a reason for hiding this comment

Uh oh!

twalthr Sep 30, 2021

Choose a reason for hiding this comment

Uh oh!

Uh oh!

flinkbot commented Sep 29, 2021 •

edited

Loading