[FLINK-25379][connectors] Support limit push down in DATAGEN connector

[ZhangChaoming]

What is the purpose of the change

Support limit push down in DATAGEN connector.

Brief change log

• Support limit push down in DATAGEN connector.

Verifying this change

Apply LIMIT in datagen table query, see DataGeneratorConnectorITCase#testLimit.

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

• Dependencies (does it add or upgrade a dependency): (yes / no)
• The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
• The serializers: (yes / no / don't know)
• The runtime per-record code paths (performance sensitive): (yes / no / don't know)
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / no / don't know)
• The S3 file system connector: (yes / no / don't know)

Documentation

• Does this pull request introduce a new feature? (yes / no)
• If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

[flinkbot]

CI report:

• 15f9429 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[flinkbot]

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 b2fc0dd (Mon Dec 20 07:18:03 UTC 2021)

Warnings:

• No documentation files were touched! Remember to keep the Flink docs up to date!
• This pull request references an unassigned Jira ticket. According to the code contribution guide, tickets need to be assigned before starting with the implementation work.

_{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

[ZhangChaoming]

@flinkbot run azure

[MartijnVisser]

You can already specify the number of records for this connector per the documentation, see https://nightlies.apache.org/flink/flink-docs-stable/docs/connectors/table/datagen/

[slinkydeveloper]

Nice change, can you add a simple test in DataGeneratorConnectorITCase that checks that using the LIMIT keyword instead of the number-of-rows option works fine?

[slinkydeveloper]

@MartijnVisser this PR adds support for the limit keyword, that is instead of defining number-of-rows per doc, you can also omit that option and instead per query set the limit like this SELECT * FROM datagentable LIMIT 10

[slinkydeveloper]

Thank you for the contribution @ZhangChaoming, just left another small comment and then it looks good!

[MartijnVisser]

I am wondering if LIMIT is the best option for the Datagen connector, given the special circumstances for this connector because it generates random rows every time. I think LIMIT implies that a user wants to limit the available (or in this case, generated) results. If I would apply a LIMIT in combination with a specific reading position for Kafka, I get the same results over and over. That's not the case for Datagen of course, because it generates random values.

I rather be consistent in what LIMIT does for all connectors and keep the current method of number-of-rows to change the connector from unbounded to bounded, so a user doesn't have to read and understand the documentation that LIMIT does something differently for Datagen compared to other connectors.

[Airblader]

The expectation that a query with LIMIT always returns the same results is only true for ordered output, which streaming queries don't support to begin with. I don't think the comparison to the Kafka connector is correct here; having to specify a fixed offset is an additional requirement (not mandated by the Kafka connector), and a similar thing could be done for datagen (specifying a seed for the RNG).

Let's keep in mind that the limit is already being applied, by Flink itself outside the source. The pushdown just allows the source itself to aid with that. I don't see a great reason why the datagen connector should need to support this, but I certainly don't see harm in it either.

[MartijnVisser]

The expectation that a query with LIMIT always returns the same results is only true for ordered output, which streaming queries don't support to begin with.

For me the question is what a user expects when its applying LIMIT. The fact we don't (yet) support it is less relevant in that context. Is there anything on LIMIT in the SQL standards?

I don't see a great reason why the datagen connector should need to support this, but I certainly don't see harm in it either.

The harm I see is that we have two options to achieve apparently the same result, which could also be conflicting with each other ("What output can I expect when I set number-of-rows to 10 and LIMIT 5")?

[sjwiesman]

@MartijnVisser

A datagen table with a set number of rows is BOUNDED and can be run under batch execution (this is why we added it originally). LIMIT is technically an orthogonal concept, but I agree it is murky.

[Airblader]

The harm I see is that we have two options to achieve apparently the same result

But LIMIT does already work, it's just not pushed into the source (for this connector). For the user nothing really changes here(?)

What output can I expect when I set number-of-rows to 10 and LIMIT 5

Why should it behave differently than for any other bounded source to which a limit is applied?

[MartijnVisser]

But LIMIT does already work, it's just not pushed into the source (for this connector). For the user nothing really changes here(?)

If that's the case, then I agree that my argument is not applicable here.

Why should it behave differently than for any other bounded source to which a limit is applied?

It shouldn't, but Datagen itself behaves differently than any other bounded source because of its randomness.

[MartijnVisser]

@ZhangChaoming Please have a look first at the CI, because the build is currently not passing

[ZhangChaoming]

@MartijnVisser I can not find any CI progress for this PR.

[MartijnVisser]

@ZhangChaoming It always updates the one from the first comment, which you can find here #18151 (comment)

[ZhangChaoming]

@flinkbot run azure

[ZhangChaoming]

@MartijnVisser Thanks, check code style failed, I will fix it.

[ZhangChaoming]

@MartijnVisser Pipeline still failed, but it seems the error was not caused by my code.

[MartijnVisser]

@ZhangChaoming You should probably rebase your branch on the latest master to resolve those issues

[ZhangChaoming]

@MartijnVisser I think my code is safe, since the error was caused by testing test_ci finegrained_resource_management. And This branch has no conflict with the base branch. Ref FLINK-26882

[MartijnVisser]

@ZhangChaoming Still, I wouldn't merge it unless CI has fully passed. Note: you'll need a review from probably @slinkydeveloper before this could be merged

[ZhangChaoming]

@flinkbot run azure

[ZhangChaoming]

@MartijnVisser The CI has passed, and could you please review this PR?

[MartijnVisser]

@ZhangChaoming I hope that @Airblader or @slinkydeveloper can take care of the review

[ZhangChaoming]

@flinkbot run azure

[ZhangChaoming]

@MartijnVisser Could you please help review this ?

[MartijnVisser]

slink

@slinkydeveloper Can you help with the review?

[MartijnVisser]

@slinkydeveloper Thanks for the review and @ZhangChaoming thanks for the PR. I've merged it.