Skip to content

[FLINK-38504][table] Add restore tests for delta join #27117

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom
Open
wants to merge 3 commits into
base: master
Choose a base branch
from
Open

[FLINK-38504][table] Add restore tests for delta join #27117

wants to merge 3 commits into from
+5,792 −37

Conversation

@xuyangzhong
Copy link
Contributor

@xuyangzhong xuyangzhong commented Oct 14, 2025 •
edited
Loading

What is the purpose of the change

Add restore tests for delta join.

Brief change log

  • add index for SourceTestStep
  • add registeredDataForFullStage in TestValuesTableFactory
  • add restore tests for delta join

Verifying this change

New tests and existent tests can verify this pr.

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

@flinkbot
Copy link
Collaborator

flinkbot commented Oct 14, 2025 •
edited
Loading

CI report:

Bot commands The @flinkbot bot supports the following commands:
  • @flinkbot run azure re-run the last Azure build

@xuyangzhong xuyangzhong force-pushed the delta_join_restore_test branch from b590638 to d899471 Compare October 27, 2025 12:54
@xuyangzhong xuyangzhong marked this pull request as ready for review October 27, 2025 12:55
@xuyangzhong xuyangzhong force-pushed the delta_join_restore_test branch 2 times, most recently from f187951 to 7026c3c Compare October 30, 2025 02:10
@xuyangzhong xuyangzhong force-pushed the delta_join_restore_test branch from 7026c3c to d2c0332 Compare October 30, 2025 06:57
Au-Miner
Au-Miner reviewed Oct 31, 2025
View reviewed changes
...ble/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/SinkTestStep.java
data.addAll(expectedAfterRestore);
}

Map<List<Object>, Row> deduplicatedMap = new HashMap<>();
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The logic in this section, which uses deduplicatedFieldIndices to modify expectedBeforeRestore and expectedAfterRestore, can also be implemented by calling a function in getExpectedBeforeRestoreAsStrings and getExpectedAfterRestoreAsStrings. Of course, it's up to you, but I personally find the latter easier to understand

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Currently, for expectedBeforeRestoreStrings and expectedAfterRestoreStrings, which have transformed each row into a string and lost their schema, it is not possible to deduplicate them by column. (Perhaps one way to handle this is by removing the leading and trailing parentheses and splitting the strings by commas).
However, for now, there is no need to support this in string scenarios for test, so I haven't implemented it yet.

...e-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java Outdated
// The difference between registeredDataForFullStage and `registeredData` is that
// `registeredData` is used for data delivered from the source to downstream, while the rows in
// `registeredDataForFullStage` will not be sent to downstream and are only used for lookup.
private static final Map<String, Collection<Row>> registeredDataForFullStage = new HashMap<>();
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm curious why it's named registeredDataForFullStage instead of ForCurrentStage or ForCurrentTable

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I have renamed it to registeredConsumedData.

Different with registeredRowData, the registeredConsumedData will not be re-consumed by the source to be sent to downstream operators.

@Au-Miner
Copy link
Contributor

Thanks for advancing the feature. Let me leave some comments!

Au-Miner
Au-Miner approved these changes Oct 31, 2025
View reviewed changes
Copy link
Contributor

@Au-Miner Au-Miner left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM, thanks.

@github-actions github-actions bot added the community-reviewed PR has been reviewed by the community. label Oct 31, 2025
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

1 more reviewer

@Au-Miner Au-Miner Au-Miner approved these changes

Reviewers whose approvals may not affect merge requirements

Assignees

No one assigned

Labels

community-reviewed PR has been reviewed by the community.

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

3 participants

@xuyangzhong @flinkbot @Au-Miner