You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2021/03/04 04:23:03 UTC
[GitHub] [flink] beyond1920 opened a new pull request #15081: [FLINK-21553][table-runtime-blink] Copy record if needed when flush window buffer records to state
beyond1920 opened a new pull request #15081:
URL: https://github.com/apache/flink/pull/15081
## What is the purpose of the change
This pull request aims to fix a minor bug in WTF which may cause unstable output.
## Brief change log
- Update CombineRecordsFunction to copy record if needed when flush window buffer records to state
- Update SlicingWindowAggOperatorBuilder to pass InputRowType to CombineRecordsFunction.Factory.
## Verifying this change
- Existed unstable case
## 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/Mesos, 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? (not applicable)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15081: [FLINK-21553][table-runtime-blink] Copy record if needed when flush window buffer records to state
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15081:
URL: https://github.com/apache/flink/pull/15081#issuecomment-790283597
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e5392081f205893b2f504a66e9ea3e9d7d0c0432",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14094",
"triggerID" : "e5392081f205893b2f504a66e9ea3e9d7d0c0432",
"triggerType" : "PUSH"
}, {
"hash" : "fd6d33709d8c33a81a29fb3352fbcd536aba6ab4",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14103",
"triggerID" : "fd6d33709d8c33a81a29fb3352fbcd536aba6ab4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* e5392081f205893b2f504a66e9ea3e9d7d0c0432 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14094)
* fd6d33709d8c33a81a29fb3352fbcd536aba6ab4 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14103)
<details>
<summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] wuchong commented on a change in pull request #15081: [FLINK-21553][table-runtime-blink] Copy record if needed when flush window buffer records to state
Posted by GitBox <gi...@apache.org>.
wuchong commented on a change in pull request #15081:
URL: https://github.com/apache/flink/pull/15081#discussion_r587210465
##########
File path: flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/WindowDistinctAggregateITCase.scala
##########
@@ -733,8 +733,9 @@ object WindowDistinctAggregateITCase {
def parameters(): util.Collection[Array[java.lang.Object]] = {
Seq[Array[AnyRef]](
Array(Boolean.box(true), HEAP_BACKEND),
+ // add SplitDistinct for HEAP back
Review comment:
remove the comment
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] beyond1920 commented on pull request #15081: [FLINK-21553][table-runtime-blink] Copy record if needed when flush window buffer records to state
Posted by GitBox <gi...@apache.org>.
beyond1920 commented on pull request #15081:
URL: https://github.com/apache/flink/pull/15081#issuecomment-790389209
@wuchong Thanks for your review, I've address all your comments.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15081: [FLINK-21553][table-runtime-blink] Copy record if needed when flush window buffer records to state
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15081:
URL: https://github.com/apache/flink/pull/15081#issuecomment-790283597
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e5392081f205893b2f504a66e9ea3e9d7d0c0432",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14094",
"triggerID" : "e5392081f205893b2f504a66e9ea3e9d7d0c0432",
"triggerType" : "PUSH"
}, {
"hash" : "fd6d33709d8c33a81a29fb3352fbcd536aba6ab4",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14103",
"triggerID" : "fd6d33709d8c33a81a29fb3352fbcd536aba6ab4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* fd6d33709d8c33a81a29fb3352fbcd536aba6ab4 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14103)
<details>
<summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] wuchong commented on a change in pull request #15081: [FLINK-21553][table-runtime-blink] Copy record if needed when flush window buffer records to state
Posted by GitBox <gi...@apache.org>.
wuchong commented on a change in pull request #15081:
URL: https://github.com/apache/flink/pull/15081#discussion_r587207786
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/aggregate/window/combines/CombineRecordsFunction.java
##########
@@ -65,21 +70,31 @@ public CombineRecordsFunction(
StateKeyContext keyContext,
WindowValueState<Long> accState,
NamespaceAggsHandleFunction<Long> aggregator,
- boolean requiresCopyKey,
+ boolean requiresCopy,
+ RowType recordType,
boolean isEventTime) {
this.timerService = timerService;
this.keyContext = keyContext;
this.accState = accState;
this.aggregator = aggregator;
- this.requiresCopyKey = requiresCopyKey;
+ this.requiresCopy = requiresCopy;
+ if (requiresCopy) {
+ LogicalType[] recordFieldTypes =
+ recordType.getFields().stream()
+ .map(RowType.RowField::getType)
+ .toArray(LogicalType[]::new);
+ this.recordSerializer = new RowDataSerializer(recordFieldTypes);
+ } else {
+ this.recordSerializer = null;
Review comment:
I think it's fine to always initialize `recordSerializer`.
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/aggregate/window/combines/CombineRecordsFunction.java
##########
@@ -152,13 +174,14 @@ public WindowCombineFunction create(
aggregator.open(
new PerWindowStateDataViewStore(
stateBackend, LongSerializer.INSTANCE, runtimeContext));
- boolean requiresCopyKey = !isStateImmutableInStateBackend(stateBackend);
+ boolean requiresCopyRecord = !isStateImmutableInStateBackend(stateBackend);
Review comment:
`requiresCopyRecord` ==> `requiresCopy`.
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/aggregate/window/combines/CombineRecordsFunction.java
##########
@@ -65,21 +70,31 @@ public CombineRecordsFunction(
StateKeyContext keyContext,
WindowValueState<Long> accState,
NamespaceAggsHandleFunction<Long> aggregator,
- boolean requiresCopyKey,
+ boolean requiresCopy,
+ RowType recordType,
boolean isEventTime) {
this.timerService = timerService;
this.keyContext = keyContext;
this.accState = accState;
this.aggregator = aggregator;
- this.requiresCopyKey = requiresCopyKey;
+ this.requiresCopy = requiresCopy;
+ if (requiresCopy) {
+ LogicalType[] recordFieldTypes =
+ recordType.getFields().stream()
+ .map(RowType.RowField::getType)
+ .toArray(LogicalType[]::new);
+ this.recordSerializer = new RowDataSerializer(recordFieldTypes);
Review comment:
Why not `new RowDataSerializer(recordType)`?
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/aggregate/window/combines/CombineRecordsFunction.java
##########
@@ -54,8 +57,10 @@
/** Function used to handle all aggregates. */
private final NamespaceAggsHandleFunction<Long> aggregator;
- /** Whether to copy input key, because key is reused. */
- private final boolean requiresCopyKey;
+ /** Whether to copy input record, because record is reused. */
Review comment:
```suggestion
/** Whether to copy key and input record, because key and record are reused. */
```
We should also mention key, because key and record are separate.
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/aggregate/window/combines/CombineRecordsFunction.java
##########
@@ -54,8 +57,10 @@
/** Function used to handle all aggregates. */
private final NamespaceAggsHandleFunction<Long> aggregator;
- /** Whether to copy input key, because key is reused. */
- private final boolean requiresCopyKey;
+ /** Whether to copy input record, because record is reused. */
+ private final boolean requiresCopy;
+
+ private final RowDataSerializer recordSerializer;
Review comment:
Add a simple Javadoc to describe it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] wuchong merged pull request #15081: [FLINK-21553][table-runtime-blink] Copy record if needed when flush window buffer records to state
Posted by GitBox <gi...@apache.org>.
wuchong merged pull request #15081:
URL: https://github.com/apache/flink/pull/15081
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15081: [FLINK-21553][table-runtime-blink] Copy record if needed when flush window buffer records to state
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15081:
URL: https://github.com/apache/flink/pull/15081#issuecomment-790283597
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e5392081f205893b2f504a66e9ea3e9d7d0c0432",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14094",
"triggerID" : "e5392081f205893b2f504a66e9ea3e9d7d0c0432",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* e5392081f205893b2f504a66e9ea3e9d7d0c0432 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14094)
<details>
<summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #15081: [FLINK-21553][table-runtime-blink] Copy record if needed when flush window buffer records to state
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #15081:
URL: https://github.com/apache/flink/pull/15081#issuecomment-790283597
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e5392081f205893b2f504a66e9ea3e9d7d0c0432",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "e5392081f205893b2f504a66e9ea3e9d7d0c0432",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* e5392081f205893b2f504a66e9ea3e9d7d0c0432 UNKNOWN
<details>
<summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #15081: [FLINK-21553][table-runtime-blink] Copy record if needed when flush window buffer records to state
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #15081:
URL: https://github.com/apache/flink/pull/15081#issuecomment-790279431
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 e5392081f205893b2f504a66e9ea3e9d7d0c0432 (Thu Mar 04 04:24:42 UTC 2021)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## 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](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
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 <summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15081: [FLINK-21553][table-runtime-blink] Copy record if needed when flush window buffer records to state
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15081:
URL: https://github.com/apache/flink/pull/15081#issuecomment-790283597
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e5392081f205893b2f504a66e9ea3e9d7d0c0432",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14094",
"triggerID" : "e5392081f205893b2f504a66e9ea3e9d7d0c0432",
"triggerType" : "PUSH"
}, {
"hash" : "fd6d33709d8c33a81a29fb3352fbcd536aba6ab4",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "fd6d33709d8c33a81a29fb3352fbcd536aba6ab4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* e5392081f205893b2f504a66e9ea3e9d7d0c0432 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14094)
* fd6d33709d8c33a81a29fb3352fbcd536aba6ab4 UNKNOWN
<details>
<summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15081: [FLINK-21553][table-runtime-blink] Copy record if needed when flush window buffer records to state
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15081:
URL: https://github.com/apache/flink/pull/15081#issuecomment-790283597
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e5392081f205893b2f504a66e9ea3e9d7d0c0432",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14094",
"triggerID" : "e5392081f205893b2f504a66e9ea3e9d7d0c0432",
"triggerType" : "PUSH"
}, {
"hash" : "fd6d33709d8c33a81a29fb3352fbcd536aba6ab4",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14103",
"triggerID" : "fd6d33709d8c33a81a29fb3352fbcd536aba6ab4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* e5392081f205893b2f504a66e9ea3e9d7d0c0432 Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14094)
* fd6d33709d8c33a81a29fb3352fbcd536aba6ab4 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14103)
<details>
<summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org