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 2020/12/01 14:55:41 UTC
[GitHub] [flink] wangxlong opened a new pull request #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
wangxlong opened a new pull request #14280:
URL: https://github.com/apache/flink/pull/14280
…nd last_value UDAF
## What is the purpose of the change
Support 'merge' method for first_value and last_value UDAF
## Brief change log
1. add merge method for first_value and last_value UDAF
2. add `valueToOrderRetractMap` and `orderToValueRetractMap` to store the retracted data.
## Verifying this change
This change added tests and can be verified as follows:
1. FirstLastValueAggFunctionWithOrderTestBase#testAggregateWithMerge
2. FirstLastValueAggFunctionWithOrderTestBase#testMergeReservedAccumulator
3. AggFunctionTestBase#testAggregateWithMerge
4. AggFunctionTestBase#testMergeReservedAccumulator
5. WindowAggregateITCase#testEventTimeSlidingWindow
## 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 / docs / JavaDocs / not documented)
----------------------------------------------------------------
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 #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736629561
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10400",
"triggerID" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"triggerType" : "PUSH"
}, {
"hash" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10424",
"triggerID" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"triggerType" : "PUSH"
}, {
"hash" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 6a0849e61034e34186780ce45dcf9157f65c0dfd Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10424)
* 57ed58c709400cabc75c5a6a4fe8587bca16dbb8 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] wangxlong commented on pull request #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
wangxlong commented on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-752823642
ping @leonardBang
----------------------------------------------------------------
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] wangxlong commented on pull request #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
wangxlong commented on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-749554417
Thanks @leonardBang for your careful review.
----------------------------------------------------------------
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 #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736629561
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10400",
"triggerID" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"triggerType" : "PUSH"
}, {
"hash" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10424",
"triggerID" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"triggerType" : "PUSH"
}, {
"hash" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10428",
"triggerID" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"triggerType" : "PUSH"
}, {
"hash" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10444",
"triggerID" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"triggerType" : "PUSH"
}, {
"hash" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11188",
"triggerID" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"triggerType" : "PUSH"
}, {
"hash" : "40095107c80dfdb13e7024ed2bb9a1760b4e41a4",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11210",
"triggerID" : "40095107c80dfdb13e7024ed2bb9a1760b4e41a4",
"triggerType" : "PUSH"
}, {
"hash" : "ac3d2a524eef87974559521807ed0defdeb6507d",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11220",
"triggerID" : "ac3d2a524eef87974559521807ed0defdeb6507d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 40095107c80dfdb13e7024ed2bb9a1760b4e41a4 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11210)
* ac3d2a524eef87974559521807ed0defdeb6507d Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11220)
<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 #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736629561
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10400",
"triggerID" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"triggerType" : "PUSH"
}, {
"hash" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10424",
"triggerID" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"triggerType" : "PUSH"
}, {
"hash" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10428",
"triggerID" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 6a0849e61034e34186780ce45dcf9157f65c0dfd Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10424)
* 57ed58c709400cabc75c5a6a4fe8587bca16dbb8 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10428)
<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 #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736629561
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10400",
"triggerID" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b2173ba7a73dc7d79ba4f0679b7076400815d116 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10400)
<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] wangxlong commented on pull request #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
wangxlong commented on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-752366742
> Could you help rebase this PR? @wangxlong The master branch merged a big format PR which caused the conflict
Sure, I will update this tonight.
----------------------------------------------------------------
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] leonardBang commented on pull request #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
leonardBang commented on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-740346160
> cc @leonardBang , could you help to review this?
ok，I’ll look into this PR
----------------------------------------------------------------
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 #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736629561
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10400",
"triggerID" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"triggerType" : "PUSH"
}, {
"hash" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10424",
"triggerID" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"triggerType" : "PUSH"
}, {
"hash" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10428",
"triggerID" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"triggerType" : "PUSH"
}, {
"hash" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10444",
"triggerID" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"triggerType" : "PUSH"
}, {
"hash" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11188",
"triggerID" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"triggerType" : "PUSH"
}, {
"hash" : "40095107c80dfdb13e7024ed2bb9a1760b4e41a4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11210",
"triggerID" : "40095107c80dfdb13e7024ed2bb9a1760b4e41a4",
"triggerType" : "PUSH"
}, {
"hash" : "ac3d2a524eef87974559521807ed0defdeb6507d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11220",
"triggerID" : "ac3d2a524eef87974559521807ed0defdeb6507d",
"triggerType" : "PUSH"
}, {
"hash" : "f03a31c7c7f46424db160faaf4daf5e9fb1b672e",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11457",
"triggerID" : "f03a31c7c7f46424db160faaf4daf5e9fb1b672e",
"triggerType" : "PUSH"
}, {
"hash" : "671c2ee902f84118152ce4d038e548a23885e208",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "671c2ee902f84118152ce4d038e548a23885e208",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f03a31c7c7f46424db160faaf4daf5e9fb1b672e Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11457)
* 671c2ee902f84118152ce4d038e548a23885e208 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 #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736629561
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10400",
"triggerID" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"triggerType" : "PUSH"
}, {
"hash" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10424",
"triggerID" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"triggerType" : "PUSH"
}, {
"hash" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10428",
"triggerID" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"triggerType" : "PUSH"
}, {
"hash" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10444",
"triggerID" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"triggerType" : "PUSH"
}, {
"hash" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11188",
"triggerID" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"triggerType" : "PUSH"
}, {
"hash" : "40095107c80dfdb13e7024ed2bb9a1760b4e41a4",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11210",
"triggerID" : "40095107c80dfdb13e7024ed2bb9a1760b4e41a4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f4861b259fafb3036cdb7a94756c6408ddfb4c32 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11188)
* 40095107c80dfdb13e7024ed2bb9a1760b4e41a4 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11210)
<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 #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736629561
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10400",
"triggerID" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"triggerType" : "PUSH"
}, {
"hash" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10424",
"triggerID" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"triggerType" : "PUSH"
}, {
"hash" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10428",
"triggerID" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"triggerType" : "PUSH"
}, {
"hash" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10444",
"triggerID" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"triggerType" : "PUSH"
}, {
"hash" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11188",
"triggerID" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"triggerType" : "PUSH"
}, {
"hash" : "40095107c80dfdb13e7024ed2bb9a1760b4e41a4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11210",
"triggerID" : "40095107c80dfdb13e7024ed2bb9a1760b4e41a4",
"triggerType" : "PUSH"
}, {
"hash" : "ac3d2a524eef87974559521807ed0defdeb6507d",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11220",
"triggerID" : "ac3d2a524eef87974559521807ed0defdeb6507d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ac3d2a524eef87974559521807ed0defdeb6507d Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11220)
<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 #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736629561
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10400",
"triggerID" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"triggerType" : "PUSH"
}, {
"hash" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10424",
"triggerID" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"triggerType" : "PUSH"
}, {
"hash" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10428",
"triggerID" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"triggerType" : "PUSH"
}, {
"hash" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10444",
"triggerID" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"triggerType" : "PUSH"
}, {
"hash" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11188",
"triggerID" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"triggerType" : "PUSH"
}, {
"hash" : "40095107c80dfdb13e7024ed2bb9a1760b4e41a4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11210",
"triggerID" : "40095107c80dfdb13e7024ed2bb9a1760b4e41a4",
"triggerType" : "PUSH"
}, {
"hash" : "ac3d2a524eef87974559521807ed0defdeb6507d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11220",
"triggerID" : "ac3d2a524eef87974559521807ed0defdeb6507d",
"triggerType" : "PUSH"
}, {
"hash" : "f03a31c7c7f46424db160faaf4daf5e9fb1b672e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11457",
"triggerID" : "f03a31c7c7f46424db160faaf4daf5e9fb1b672e",
"triggerType" : "PUSH"
}, {
"hash" : "671c2ee902f84118152ce4d038e548a23885e208",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11491",
"triggerID" : "671c2ee902f84118152ce4d038e548a23885e208",
"triggerType" : "PUSH"
}, {
"hash" : "bdbf525b9e6435c9e6c9666c1c1ba0cf88ffff9a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "bdbf525b9e6435c9e6c9666c1c1ba0cf88ffff9a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 671c2ee902f84118152ce4d038e548a23885e208 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11491)
* bdbf525b9e6435c9e6c9666c1c1ba0cf88ffff9a 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 #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736629561
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10400",
"triggerID" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"triggerType" : "PUSH"
}, {
"hash" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10424",
"triggerID" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"triggerType" : "PUSH"
}, {
"hash" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10428",
"triggerID" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"triggerType" : "PUSH"
}, {
"hash" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 57ed58c709400cabc75c5a6a4fe8587bca16dbb8 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10428)
* 851970c1c94f8f44058f914d3dbeb959cb66ee53 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] wuchong commented on pull request #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
wuchong commented on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-740345760
cc @leonardBang , could you help to review this?
----------------------------------------------------------------
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 #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736629561
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10400",
"triggerID" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"triggerType" : "PUSH"
}, {
"hash" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10424",
"triggerID" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"triggerType" : "PUSH"
}, {
"hash" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10428",
"triggerID" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"triggerType" : "PUSH"
}, {
"hash" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10444",
"triggerID" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"triggerType" : "PUSH"
}, {
"hash" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11188",
"triggerID" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"triggerType" : "PUSH"
}, {
"hash" : "40095107c80dfdb13e7024ed2bb9a1760b4e41a4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11210",
"triggerID" : "40095107c80dfdb13e7024ed2bb9a1760b4e41a4",
"triggerType" : "PUSH"
}, {
"hash" : "ac3d2a524eef87974559521807ed0defdeb6507d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11220",
"triggerID" : "ac3d2a524eef87974559521807ed0defdeb6507d",
"triggerType" : "PUSH"
}, {
"hash" : "f03a31c7c7f46424db160faaf4daf5e9fb1b672e",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11457",
"triggerID" : "f03a31c7c7f46424db160faaf4daf5e9fb1b672e",
"triggerType" : "PUSH"
}, {
"hash" : "671c2ee902f84118152ce4d038e548a23885e208",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11491",
"triggerID" : "671c2ee902f84118152ce4d038e548a23885e208",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f03a31c7c7f46424db160faaf4daf5e9fb1b672e Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11457)
* 671c2ee902f84118152ce4d038e548a23885e208 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11491)
<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] JingsongLi commented on pull request #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-862185730
Close the inactive PR. Feel free to re-open.
--
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 #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736629561
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10400",
"triggerID" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"triggerType" : "PUSH"
}, {
"hash" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10424",
"triggerID" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"triggerType" : "PUSH"
}, {
"hash" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10428",
"triggerID" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"triggerType" : "PUSH"
}, {
"hash" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10444",
"triggerID" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"triggerType" : "PUSH"
}, {
"hash" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11188",
"triggerID" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 851970c1c94f8f44058f914d3dbeb959cb66ee53 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10444)
* f4861b259fafb3036cdb7a94756c6408ddfb4c32 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11188)
<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 #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736606688
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 bdbf525b9e6435c9e6c9666c1c1ba0cf88ffff9a (Sat Aug 28 12:17:14 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](https://issues.apache.org/jira/browse/FLINK-20110).** According to the [code contribution guide](https://flink.apache.org/contributing/contribute-code.html), tickets need to be assigned before starting with the implementation work.
<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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736629561
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10400",
"triggerID" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"triggerType" : "PUSH"
}, {
"hash" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10424",
"triggerID" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"triggerType" : "PUSH"
}, {
"hash" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10428",
"triggerID" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"triggerType" : "PUSH"
}, {
"hash" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10444",
"triggerID" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"triggerType" : "PUSH"
}, {
"hash" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11188",
"triggerID" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"triggerType" : "PUSH"
}, {
"hash" : "40095107c80dfdb13e7024ed2bb9a1760b4e41a4",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11210",
"triggerID" : "40095107c80dfdb13e7024ed2bb9a1760b4e41a4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 40095107c80dfdb13e7024ed2bb9a1760b4e41a4 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11210)
<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] leonardBang commented on a change in pull request #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
leonardBang commented on a change in pull request #14280:
URL: https://github.com/apache/flink/pull/14280#discussion_r555055866
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/FirstValueAggFunction.java
##########
@@ -83,7 +83,7 @@ public void accumulate(RowData rowData, Object value) {
GenericRowData acc = (GenericRowData) rowData;
if (value != null && acc.getLong(1) == Long.MAX_VALUE) {
acc.setField(0, value);
- acc.setField(1, System.currentTimeMillis());
+ acc.setField(1, System.nanoTime());
Review comment:
The `System.nanoTime()` has higher precision but it can not be used in different JVMs because every JVM because there's no guarantee that nanoTime is relative to any timebase.
----------------------------------------------------------------
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 #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736629561
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10400",
"triggerID" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"triggerType" : "PUSH"
}, {
"hash" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10424",
"triggerID" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"triggerType" : "PUSH"
}, {
"hash" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10428",
"triggerID" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"triggerType" : "PUSH"
}, {
"hash" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10444",
"triggerID" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 851970c1c94f8f44058f914d3dbeb959cb66ee53 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10444)
<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] leonardBang commented on a change in pull request #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
leonardBang commented on a change in pull request #14280:
URL: https://github.com/apache/flink/pull/14280#discussion_r546504664
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/FirstValueAggFunction.java
##########
@@ -107,6 +107,27 @@ public void accumulate(RowData rowData, StringData value, Long order) {
}
}
+ public void merge(RowData acc, Iterable<RowData> its) {
Review comment:
`its ` => `accIt` ?
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/LastValueAggFunction.java
##########
@@ -106,6 +107,27 @@ public void accumulate(GenericRowData acc, StringData value, Long order) {
}
}
+ public void merge(RowData acc, Iterable<RowData> its) {
Review comment:
`its` => `accIt`
----------------------------------------------------------------
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 #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736629561
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10400",
"triggerID" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"triggerType" : "PUSH"
}, {
"hash" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10424",
"triggerID" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"triggerType" : "PUSH"
}, {
"hash" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10428",
"triggerID" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"triggerType" : "PUSH"
}, {
"hash" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10444",
"triggerID" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 57ed58c709400cabc75c5a6a4fe8587bca16dbb8 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10428)
* 851970c1c94f8f44058f914d3dbeb959cb66ee53 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10444)
<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] leonardBang commented on a change in pull request #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
leonardBang commented on a change in pull request #14280:
URL: https://github.com/apache/flink/pull/14280#discussion_r546504664
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/FirstValueAggFunction.java
##########
@@ -107,6 +107,27 @@ public void accumulate(RowData rowData, StringData value, Long order) {
}
}
+ public void merge(RowData acc, Iterable<RowData> its) {
Review comment:
`its ` => `it` ?
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/LastValueAggFunction.java
##########
@@ -106,6 +107,27 @@ public void accumulate(GenericRowData acc, StringData value, Long order) {
}
}
+ public void merge(RowData acc, Iterable<RowData> its) {
Review comment:
`its` => `it`
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/FirstValueAggFunction.java
##########
@@ -83,7 +83,7 @@ public void accumulate(RowData rowData, Object value) {
GenericRowData acc = (GenericRowData) rowData;
if (value != null && acc.getLong(1) == Long.MAX_VALUE) {
acc.setField(0, value);
- acc.setField(1, System.currentTimeMillis());
+ acc.setField(1, System.nanoTime());
Review comment:
I think we'd better use `System.currentTimeMillis()` which is matched with Flink PROC_TIME semantics, and the `System.nanoTime()` cannot represent current timestamp, it's just a relative time with higher precision.
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/LastValueAggFunction.java
##########
@@ -106,6 +107,27 @@ public void accumulate(GenericRowData acc, StringData value, Long order) {
}
}
+ public void merge(RowData acc, Iterable<RowData> its) {
+ GenericRowData accRowData = (GenericRowData) acc;
+ long accOrder = accRowData.getLong(1);
+ GenericRowData firstAcc = accRowData;
Review comment:
please rename the variable
```suggestion
GenericRowData lastAcc = accRowData;
```
##########
File path: flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/functions/aggfunctions/AggFunctionTestBase.java
##########
@@ -100,15 +100,15 @@ public void testAggregateWithMerge()
List<T> inputValues = inputValueSets.get(i);
T expected = expectedResults.get(i);
// equally split the vals sequence into two sequences
- Tuple2<List<T>, List<T>> splitValues = splitValues(inputValues);
+ Tuple2<List, List> splitValues = splitValues(inputValues);
List<T> firstValues = splitValues.f0;
List<T> secondValues = splitValues.f1;
+
// 1. verify merge with accumulate
+ ACC acc = accumulateValues(firstValues);
List<ACC> accumulators = new ArrayList<>();
accumulators.add(accumulateValues(secondValues));
- ACC acc = accumulateValues(firstValues);
-
Review comment:
I think the `<T>` is meaningful, and we shouldn't update these code
##########
File path: flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/functions/aggfunctions/FirstLastValueAggFunctionWithOrderTestBase.java
##########
@@ -73,6 +76,112 @@ public void testAccumulateAndRetractWithoutMerge()
}
}
+ @Test
+ public void testAggregateWithMerge()
+ throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {
+ AggregateFunction<T, ACC> aggregator = getAggregator();
+ if (UserDefinedFunctionUtils.ifMethodExistInFunction("merge", aggregator)) {
+ Method mergeFunc = aggregator.getClass().getMethod("merge", getAccClass(), Iterable.class);
+ List<List<T>> inputValueSets = getInputValueSets();
+ List<List<Long>> inputOrderSets = getInputOrderSets();
+ List<T> expectedResults = getExpectedResults();
+ Preconditions.checkArgument(inputValueSets.size() == expectedResults.size());
+ int size = getInputValueSets().size();
+ // iterate over input sets
+ for (int i = 0; i < size; ++i) {
+ List<T> inputValues = inputValueSets.get(i);
+ List<Long> inputOrders = inputOrderSets.get(i);
+ T expected = expectedResults.get(i);
+ // equally split the vals sequence into two sequences
+ Tuple2<List, List> splitValues = splitValues(inputValues);
+ Tuple2<List, List> splitOrders = splitValues(inputOrders);
Review comment:
minor: Using `Tuple2<List<T>, List<T>>` to eliminate IDE compile warings
##########
File path: flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/WindowAggregateITCase.scala
##########
@@ -77,7 +77,9 @@ class WindowAggregateITCase(mode: StateBackendMode)
| SUM(1),
| COUNT(`int`),
| COUNT(DISTINCT `float`),
- | concat_distinct_agg(name)
+ | concat_distinct_agg(name),
+ | first_value(`int`),
Review comment:
Can we also add an IT test for first_value/last_value with retraction?
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/FirstLastValueWithRetractAggFunctionBase.java
##########
@@ -0,0 +1,483 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.functions.aggfunctions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.dataview.MapView;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.binary.BinaryStringData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.BiFunction;
+
+import static org.apache.flink.table.types.utils.DataTypeUtils.toInternalDataType;
+
+/**
+ * Base retraction aggregate function for FIRST_VALUE and LAST_VALUE.
+ */
+@Internal
+abstract class FirstLastValueWithRetractAggFunctionBase<T>
+ extends InternalAggregateFunction<T, FirstLastValueWithRetractAggFunctionBase.FirstLastValueWithRetractAccumulator<T>> {
+
+ private transient DataType valueDataType;
+
+ public FirstLastValueWithRetractAggFunctionBase(LogicalType valueType) {
+ this.valueDataType = toInternalDataType(valueType);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Planning
+ // --------------------------------------------------------------------------------------------
+
+ @Override
+ public DataType[] getInputDataTypes() {
+ return new DataType[]{valueDataType};
+ }
+
+ @Override
+ public DataType getAccumulatorDataType() {
+ return DataTypes.STRUCTURED(
+ FirstLastValueWithRetractAccumulator.class,
+ DataTypes.FIELD(
+ "value",
+ valueDataType.nullable()),
+ DataTypes.FIELD(
+ "order",
+ DataTypes.BIGINT()),
+ DataTypes.FIELD(
+ "valueToOrderMap",
+ MapView.newMapViewDataType(
+ valueDataType.notNull(),
+ DataTypes.ARRAY(DataTypes.BIGINT()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "valueToOrderRetractMap",
+ MapView.newMapViewDataType(
+ valueDataType.notNull(),
+ DataTypes.ARRAY(DataTypes.BIGINT()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "orderToValueMap",
+ MapView.newMapViewDataType(
+ DataTypes.BIGINT(),
+ DataTypes.ARRAY(valueDataType.notNull()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "orderToValueRetractMap",
+ MapView.newMapViewDataType(
+ DataTypes.BIGINT(),
+ DataTypes.ARRAY(valueDataType.notNull()).bridgedTo(List.class)))
+ );
+ }
+
+ @Override
+ public DataType getOutputDataType() {
+ return valueDataType;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Runtime
+ // --------------------------------------------------------------------------------------------
+
+ /** Accumulator for FIRST_VALUE or LAST_VALUE. */
Review comment:
or -> and
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/FirstLastValueWithRetractAggFunctionBase.java
##########
@@ -0,0 +1,483 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.functions.aggfunctions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.dataview.MapView;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.binary.BinaryStringData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.BiFunction;
+
+import static org.apache.flink.table.types.utils.DataTypeUtils.toInternalDataType;
+
+/**
+ * Base retraction aggregate function for FIRST_VALUE and LAST_VALUE.
+ */
+@Internal
+abstract class FirstLastValueWithRetractAggFunctionBase<T>
+ extends InternalAggregateFunction<T, FirstLastValueWithRetractAggFunctionBase.FirstLastValueWithRetractAccumulator<T>> {
+
+ private transient DataType valueDataType;
+
+ public FirstLastValueWithRetractAggFunctionBase(LogicalType valueType) {
+ this.valueDataType = toInternalDataType(valueType);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Planning
+ // --------------------------------------------------------------------------------------------
+
+ @Override
+ public DataType[] getInputDataTypes() {
+ return new DataType[]{valueDataType};
+ }
+
+ @Override
+ public DataType getAccumulatorDataType() {
+ return DataTypes.STRUCTURED(
+ FirstLastValueWithRetractAccumulator.class,
+ DataTypes.FIELD(
+ "value",
+ valueDataType.nullable()),
+ DataTypes.FIELD(
+ "order",
+ DataTypes.BIGINT()),
+ DataTypes.FIELD(
+ "valueToOrderMap",
+ MapView.newMapViewDataType(
+ valueDataType.notNull(),
+ DataTypes.ARRAY(DataTypes.BIGINT()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "valueToOrderRetractMap",
+ MapView.newMapViewDataType(
+ valueDataType.notNull(),
+ DataTypes.ARRAY(DataTypes.BIGINT()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "orderToValueMap",
+ MapView.newMapViewDataType(
+ DataTypes.BIGINT(),
+ DataTypes.ARRAY(valueDataType.notNull()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "orderToValueRetractMap",
+ MapView.newMapViewDataType(
+ DataTypes.BIGINT(),
+ DataTypes.ARRAY(valueDataType.notNull()).bridgedTo(List.class)))
+ );
+ }
+
+ @Override
+ public DataType getOutputDataType() {
+ return valueDataType;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Runtime
+ // --------------------------------------------------------------------------------------------
+
+ /** Accumulator for FIRST_VALUE or LAST_VALUE. */
+ public static class FirstLastValueWithRetractAccumulator<T> {
+ public T value = null;
+ public Long order = null;
+ public MapView<T, List<Long>> valueToOrderMap = new MapView<>();
+ public MapView<T, List<Long>> valueToOrderRetractMap = new MapView<>();
+ public MapView<Long, List<T>> orderToValueMap = new MapView<>();
+ public MapView<Long, List<T>> orderToValueRetractMap = new MapView<>();
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof FirstLastValueWithRetractAccumulator)) {
+ return false;
+ }
+ FirstLastValueWithRetractAccumulator<?> that = (FirstLastValueWithRetractAccumulator<?>) o;
+ return Objects.equals(value, that.value) &&
+ Objects.equals(order, that.order) &&
+ Objects.equals(valueToOrderMap, that.valueToOrderMap) &&
+ Objects.equals(valueToOrderRetractMap, that.valueToOrderRetractMap) &&
+ Objects.equals(orderToValueMap, that.orderToValueMap) &&
+ Objects.equals(orderToValueRetractMap, that.orderToValueRetractMap);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ value,
+ order,
+ valueToOrderMap,
+ valueToOrderRetractMap,
+ orderToValueMap,
+ orderToValueRetractMap);
+ }
+ }
+
+ @Override
+ public FirstLastValueWithRetractAccumulator<T> createAccumulator() {
+ return new FirstLastValueWithRetractAccumulator<>();
+ }
+
+ @SuppressWarnings("unchecked")
+ public void accumulate(
+ FirstLastValueWithRetractAccumulator<T> acc,
+ Object value,
+ BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ Long order = System.nanoTime();
+ accumulate(acc, v, order, compare);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void accumulate(
+ FirstLastValueWithRetractAccumulator<T> acc,
+ Object value, Long order,
+ BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ Long prevOrder = acc.order;
+ if (prevOrder == null || compare.apply(prevOrder, order)) {
+ acc.value = v;
+ acc.order = order;
+ }
+
+ // save order to orderToValueMap
Review comment:
```suggestion
// save order to valueToOrderMap
```
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/FirstLastValueWithRetractAggFunctionBase.java
##########
@@ -0,0 +1,483 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.functions.aggfunctions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.dataview.MapView;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.binary.BinaryStringData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.BiFunction;
+
+import static org.apache.flink.table.types.utils.DataTypeUtils.toInternalDataType;
+
+/**
+ * Base retraction aggregate function for FIRST_VALUE and LAST_VALUE.
+ */
+@Internal
+abstract class FirstLastValueWithRetractAggFunctionBase<T>
+ extends InternalAggregateFunction<T, FirstLastValueWithRetractAggFunctionBase.FirstLastValueWithRetractAccumulator<T>> {
+
+ private transient DataType valueDataType;
+
+ public FirstLastValueWithRetractAggFunctionBase(LogicalType valueType) {
+ this.valueDataType = toInternalDataType(valueType);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Planning
+ // --------------------------------------------------------------------------------------------
+
+ @Override
+ public DataType[] getInputDataTypes() {
+ return new DataType[]{valueDataType};
+ }
+
+ @Override
+ public DataType getAccumulatorDataType() {
+ return DataTypes.STRUCTURED(
+ FirstLastValueWithRetractAccumulator.class,
+ DataTypes.FIELD(
+ "value",
+ valueDataType.nullable()),
+ DataTypes.FIELD(
+ "order",
+ DataTypes.BIGINT()),
+ DataTypes.FIELD(
+ "valueToOrderMap",
+ MapView.newMapViewDataType(
+ valueDataType.notNull(),
+ DataTypes.ARRAY(DataTypes.BIGINT()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "valueToOrderRetractMap",
+ MapView.newMapViewDataType(
+ valueDataType.notNull(),
+ DataTypes.ARRAY(DataTypes.BIGINT()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "orderToValueMap",
+ MapView.newMapViewDataType(
+ DataTypes.BIGINT(),
+ DataTypes.ARRAY(valueDataType.notNull()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "orderToValueRetractMap",
+ MapView.newMapViewDataType(
+ DataTypes.BIGINT(),
+ DataTypes.ARRAY(valueDataType.notNull()).bridgedTo(List.class)))
+ );
+ }
+
+ @Override
+ public DataType getOutputDataType() {
+ return valueDataType;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Runtime
+ // --------------------------------------------------------------------------------------------
+
+ /** Accumulator for FIRST_VALUE or LAST_VALUE. */
+ public static class FirstLastValueWithRetractAccumulator<T> {
+ public T value = null;
+ public Long order = null;
+ public MapView<T, List<Long>> valueToOrderMap = new MapView<>();
+ public MapView<T, List<Long>> valueToOrderRetractMap = new MapView<>();
+ public MapView<Long, List<T>> orderToValueMap = new MapView<>();
+ public MapView<Long, List<T>> orderToValueRetractMap = new MapView<>();
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof FirstLastValueWithRetractAccumulator)) {
+ return false;
+ }
+ FirstLastValueWithRetractAccumulator<?> that = (FirstLastValueWithRetractAccumulator<?>) o;
+ return Objects.equals(value, that.value) &&
+ Objects.equals(order, that.order) &&
+ Objects.equals(valueToOrderMap, that.valueToOrderMap) &&
+ Objects.equals(valueToOrderRetractMap, that.valueToOrderRetractMap) &&
+ Objects.equals(orderToValueMap, that.orderToValueMap) &&
+ Objects.equals(orderToValueRetractMap, that.orderToValueRetractMap);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ value,
+ order,
+ valueToOrderMap,
+ valueToOrderRetractMap,
+ orderToValueMap,
+ orderToValueRetractMap);
+ }
+ }
+
+ @Override
+ public FirstLastValueWithRetractAccumulator<T> createAccumulator() {
+ return new FirstLastValueWithRetractAccumulator<>();
+ }
+
+ @SuppressWarnings("unchecked")
+ public void accumulate(
+ FirstLastValueWithRetractAccumulator<T> acc,
+ Object value,
+ BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ Long order = System.nanoTime();
+ accumulate(acc, v, order, compare);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void accumulate(
+ FirstLastValueWithRetractAccumulator<T> acc,
+ Object value, Long order,
+ BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ Long prevOrder = acc.order;
+ if (prevOrder == null || compare.apply(prevOrder, order)) {
+ acc.value = v;
+ acc.order = order;
+ }
+
+ // save order to orderToValueMap
+ List<Long> orderList = acc.valueToOrderMap.get(v);
+ if (orderList == null) {
+ orderList = new ArrayList<>();
+ orderList.add(order);
+ } else {
+ // use insertion sort,
+ // for the O(1) time complexity if order is asc, compare from back to front
+ int i = orderList.size();
+ boolean isInsert = false;
+ while (i > 0) {
+ Long tempOrder = orderList.get(i - 1);
+ if (tempOrder <= order) {
+ orderList.add(i, order);
+ isInsert = true;
+ break;
+ }
+ i--;
+ }
+ if (!isInsert) {
+ orderList.add(0, order);
+ }
+ }
+ // update valueToOrderMap
+ acc.valueToOrderMap.put(v, orderList);
+
+ // update orderToValueMap
+ List<T> valueList = acc.orderToValueMap.get(order);
+ if (valueList == null) {
+ valueList = new ArrayList<>();
+ }
+ valueList.add(v);
+ acc.orderToValueMap.put(order, valueList);
+ }
+ }
+
+ public void accumulate(FirstLastValueWithRetractAccumulator<T> acc, StringData value, BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ accumulate(acc, (Object) ((BinaryStringData) value).copy(), compare);
+ }
+ }
+
+ public void accumulate(FirstLastValueWithRetractAccumulator<T> acc, StringData value, Long order, BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ accumulate(acc, (Object) ((BinaryStringData) value).copy(), order, compare);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void retract(FirstLastValueWithRetractAccumulator<T> acc, Object value) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ List<Long> orderList = acc.valueToOrderMap.get(v);
+ if (orderList != null && orderList.size() > 0) {
+ Long order = orderList.get(0);
+ retract(acc, v, order);
+ }
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void retract(FirstLastValueWithRetractAccumulator<T> acc, Object value, Long order) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ // remove from valueToOrderMap
+ List<Long> orderList = acc.valueToOrderMap.get(v);
+ if (orderList != null && orderList.size() > 0) {
+ orderList.remove(0);
+ if (orderList.isEmpty()) {
+ acc.valueToOrderMap.remove(v);
+ } else {
+ acc.valueToOrderMap.put(v, orderList);
+ }
+ } else {
+ // if not exist, save to valueToOrderRetractMap
+ List<Long> retractOrderList = acc.valueToOrderRetractMap.get(v);
+ if (retractOrderList == null) {
+ retractOrderList = new ArrayList<>();
+ }
+ retractOrderList.add(order);
+ acc.valueToOrderRetractMap.put(v, retractOrderList);
+ }
+ // remove from orderToValueMap
+ List<T> valueList = acc.orderToValueMap.get(order);
+ int index = -1;
+ if (valueList != null && (index = valueList.indexOf(v)) >= 0) {
+ valueList.remove(index);
+ if (valueList.isEmpty()) {
+ acc.orderToValueMap.remove(order);
+ } else {
+ acc.orderToValueMap.put(order, valueList);
+ }
+ } else {
+ // if not exist, save to orderToValueRetractMap
+ List<T> retractValueList = acc.orderToValueRetractMap.get(order);
+ if (retractValueList == null) {
+ retractValueList = new ArrayList<>();
+ }
+ retractValueList.add(v);
+ acc.orderToValueRetractMap.put(order, retractValueList);
+ }
+ // if retract current first value
Review comment:
```suggestion
// update the acc value if the retracted value equals first/last value
```
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/FirstLastValueWithRetractAggFunctionBase.java
##########
@@ -0,0 +1,483 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.functions.aggfunctions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.dataview.MapView;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.binary.BinaryStringData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.BiFunction;
+
+import static org.apache.flink.table.types.utils.DataTypeUtils.toInternalDataType;
+
+/**
+ * Base retraction aggregate function for FIRST_VALUE and LAST_VALUE.
+ */
+@Internal
+abstract class FirstLastValueWithRetractAggFunctionBase<T>
+ extends InternalAggregateFunction<T, FirstLastValueWithRetractAggFunctionBase.FirstLastValueWithRetractAccumulator<T>> {
+
+ private transient DataType valueDataType;
+
+ public FirstLastValueWithRetractAggFunctionBase(LogicalType valueType) {
+ this.valueDataType = toInternalDataType(valueType);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Planning
+ // --------------------------------------------------------------------------------------------
+
+ @Override
+ public DataType[] getInputDataTypes() {
+ return new DataType[]{valueDataType};
+ }
+
+ @Override
+ public DataType getAccumulatorDataType() {
+ return DataTypes.STRUCTURED(
+ FirstLastValueWithRetractAccumulator.class,
+ DataTypes.FIELD(
+ "value",
+ valueDataType.nullable()),
+ DataTypes.FIELD(
+ "order",
+ DataTypes.BIGINT()),
+ DataTypes.FIELD(
+ "valueToOrderMap",
+ MapView.newMapViewDataType(
+ valueDataType.notNull(),
+ DataTypes.ARRAY(DataTypes.BIGINT()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "valueToOrderRetractMap",
+ MapView.newMapViewDataType(
+ valueDataType.notNull(),
+ DataTypes.ARRAY(DataTypes.BIGINT()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "orderToValueMap",
+ MapView.newMapViewDataType(
+ DataTypes.BIGINT(),
+ DataTypes.ARRAY(valueDataType.notNull()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "orderToValueRetractMap",
+ MapView.newMapViewDataType(
+ DataTypes.BIGINT(),
+ DataTypes.ARRAY(valueDataType.notNull()).bridgedTo(List.class)))
+ );
+ }
+
+ @Override
+ public DataType getOutputDataType() {
+ return valueDataType;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Runtime
+ // --------------------------------------------------------------------------------------------
+
+ /** Accumulator for FIRST_VALUE or LAST_VALUE. */
+ public static class FirstLastValueWithRetractAccumulator<T> {
+ public T value = null;
+ public Long order = null;
+ public MapView<T, List<Long>> valueToOrderMap = new MapView<>();
+ public MapView<T, List<Long>> valueToOrderRetractMap = new MapView<>();
+ public MapView<Long, List<T>> orderToValueMap = new MapView<>();
+ public MapView<Long, List<T>> orderToValueRetractMap = new MapView<>();
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof FirstLastValueWithRetractAccumulator)) {
+ return false;
+ }
+ FirstLastValueWithRetractAccumulator<?> that = (FirstLastValueWithRetractAccumulator<?>) o;
+ return Objects.equals(value, that.value) &&
+ Objects.equals(order, that.order) &&
+ Objects.equals(valueToOrderMap, that.valueToOrderMap) &&
+ Objects.equals(valueToOrderRetractMap, that.valueToOrderRetractMap) &&
+ Objects.equals(orderToValueMap, that.orderToValueMap) &&
+ Objects.equals(orderToValueRetractMap, that.orderToValueRetractMap);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ value,
+ order,
+ valueToOrderMap,
+ valueToOrderRetractMap,
+ orderToValueMap,
+ orderToValueRetractMap);
+ }
+ }
+
+ @Override
+ public FirstLastValueWithRetractAccumulator<T> createAccumulator() {
+ return new FirstLastValueWithRetractAccumulator<>();
+ }
+
+ @SuppressWarnings("unchecked")
+ public void accumulate(
+ FirstLastValueWithRetractAccumulator<T> acc,
+ Object value,
+ BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ Long order = System.nanoTime();
+ accumulate(acc, v, order, compare);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void accumulate(
+ FirstLastValueWithRetractAccumulator<T> acc,
+ Object value, Long order,
+ BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ Long prevOrder = acc.order;
+ if (prevOrder == null || compare.apply(prevOrder, order)) {
+ acc.value = v;
+ acc.order = order;
+ }
+
+ // save order to orderToValueMap
+ List<Long> orderList = acc.valueToOrderMap.get(v);
+ if (orderList == null) {
+ orderList = new ArrayList<>();
+ orderList.add(order);
+ } else {
+ // use insertion sort,
+ // for the O(1) time complexity if order is asc, compare from back to front
+ int i = orderList.size();
+ boolean isInsert = false;
+ while (i > 0) {
+ Long tempOrder = orderList.get(i - 1);
+ if (tempOrder <= order) {
+ orderList.add(i, order);
+ isInsert = true;
+ break;
+ }
+ i--;
+ }
+ if (!isInsert) {
+ orderList.add(0, order);
+ }
+ }
+ // update valueToOrderMap
+ acc.valueToOrderMap.put(v, orderList);
+
+ // update orderToValueMap
+ List<T> valueList = acc.orderToValueMap.get(order);
+ if (valueList == null) {
+ valueList = new ArrayList<>();
+ }
+ valueList.add(v);
+ acc.orderToValueMap.put(order, valueList);
+ }
+ }
+
+ public void accumulate(FirstLastValueWithRetractAccumulator<T> acc, StringData value, BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ accumulate(acc, (Object) ((BinaryStringData) value).copy(), compare);
+ }
+ }
+
+ public void accumulate(FirstLastValueWithRetractAccumulator<T> acc, StringData value, Long order, BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ accumulate(acc, (Object) ((BinaryStringData) value).copy(), order, compare);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void retract(FirstLastValueWithRetractAccumulator<T> acc, Object value) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ List<Long> orderList = acc.valueToOrderMap.get(v);
+ if (orderList != null && orderList.size() > 0) {
+ Long order = orderList.get(0);
+ retract(acc, v, order);
+ }
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void retract(FirstLastValueWithRetractAccumulator<T> acc, Object value, Long order) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ // remove from valueToOrderMap
+ List<Long> orderList = acc.valueToOrderMap.get(v);
+ if (orderList != null && orderList.size() > 0) {
+ orderList.remove(0);
+ if (orderList.isEmpty()) {
+ acc.valueToOrderMap.remove(v);
+ } else {
+ acc.valueToOrderMap.put(v, orderList);
+ }
+ } else {
+ // if not exist, save to valueToOrderRetractMap
+ List<Long> retractOrderList = acc.valueToOrderRetractMap.get(v);
+ if (retractOrderList == null) {
+ retractOrderList = new ArrayList<>();
+ }
+ retractOrderList.add(order);
+ acc.valueToOrderRetractMap.put(v, retractOrderList);
+ }
+ // remove from orderToValueMap
+ List<T> valueList = acc.orderToValueMap.get(order);
+ int index = -1;
+ if (valueList != null && (index = valueList.indexOf(v)) >= 0) {
+ valueList.remove(index);
+ if (valueList.isEmpty()) {
+ acc.orderToValueMap.remove(order);
+ } else {
+ acc.orderToValueMap.put(order, valueList);
+ }
+ } else {
+ // if not exist, save to orderToValueRetractMap
+ List<T> retractValueList = acc.orderToValueRetractMap.get(order);
+ if (retractValueList == null) {
+ retractValueList = new ArrayList<>();
+ }
+ retractValueList.add(v);
+ acc.orderToValueRetractMap.put(order, retractValueList);
+ }
+ // if retract current first value
+ if (v.equals(acc.value)) {
+ updateValue(acc);
+ }
+ }
+ }
+
+ public void merge(
+ FirstLastValueWithRetractAccumulator<T> acc,
+ Iterable<FirstLastValueWithRetractAccumulator<T>> its,
+ BiFunction<Long, Long, Boolean> compare) throws Exception {
+ for (FirstLastValueWithRetractAccumulator<T> it : its) {
Review comment:
```suggestion
for (FirstLastValueWithRetractAccumulator<T> otherAcc : accIt) {
```
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/LastValueAggFunction.java
##########
@@ -83,6 +83,7 @@ public void accumulate(RowData rowData, Object value) {
GenericRowData acc = (GenericRowData) rowData;
if (value != null) {
acc.setField(0, value);
+ acc.setField(1, System.nanoTime());
Review comment:
ditto
##########
File path: flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/functions/aggfunctions/AggFunctionTestBase.java
##########
@@ -240,13 +240,13 @@ protected void retractValues(ACC accumulator, List<T> values)
}
}
- protected Tuple2<List<T>, List<T>> splitValues(List<T> values) {
+ protected Tuple2<List, List> splitValues(List values) {
return splitValues(values, values.size() / 2);
}
- protected Tuple2<List<T>, List<T>> splitValues(List<T> values, int index) {
- List<T> firstValues = new ArrayList<>();
- List<T> secondValues = new ArrayList<>();
+ protected Tuple2<List, List> splitValues(List values, int index) {
+ List firstValues = new ArrayList<>();
+ List secondValues = new ArrayList<>();
Review comment:
Could we revert this change?
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/LastValueAggFunction.java
##########
@@ -106,6 +107,27 @@ public void accumulate(GenericRowData acc, StringData value, Long order) {
}
}
+ public void merge(RowData acc, Iterable<RowData> its) {
+ GenericRowData accRowData = (GenericRowData) acc;
+ long accOrder = accRowData.getLong(1);
+ GenericRowData firstAcc = accRowData;
+ boolean needUpdate = false;
+
+ for (RowData rowData : its) {
+ long order = rowData.getLong(1);
+ // if there is a upper order
+ if (accOrder < order) {
Review comment:
Minor: We can exchange the operands order to make code more readable
```suggestion
if (order > accOrder) {
```
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/FirstValueAggFunction.java
##########
@@ -107,6 +107,27 @@ public void accumulate(RowData rowData, StringData value, Long order) {
}
}
+ public void merge(RowData acc, Iterable<RowData> its) {
+ GenericRowData accRowData = (GenericRowData) acc;
+ long accOrder = accRowData.getLong(1);
+ GenericRowData firstAcc = accRowData;
+ boolean needUpdate = false;
+
+ for (RowData rowData : its) {
+ long order = rowData.getLong(1);
+ // if there is a lower order
+ if (accOrder > order) {
Review comment:
minor: We can exchange the operands order to make code more readable
```suggestion
if (order < accOrder) {
```
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/FirstValueAggFunction.java
##########
@@ -107,6 +107,27 @@ public void accumulate(RowData rowData, StringData value, Long order) {
}
}
+ public void merge(RowData acc, Iterable<RowData> its) {
+ GenericRowData accRowData = (GenericRowData) acc;
+ long accOrder = accRowData.getLong(1);
+ GenericRowData firstAcc = accRowData;
+ boolean needUpdate = false;
+
+ for (RowData rowData : its) {
Review comment:
How about use Iterator ?
```
Iterator<RowData> iterator = it.iterator();
while (iterator.hasNext()) {
```
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/FirstLastValueWithRetractAggFunctionBase.java
##########
@@ -0,0 +1,483 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.functions.aggfunctions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.dataview.MapView;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.binary.BinaryStringData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.BiFunction;
+
+import static org.apache.flink.table.types.utils.DataTypeUtils.toInternalDataType;
+
+/**
+ * Base retraction aggregate function for FIRST_VALUE and LAST_VALUE.
Review comment:
minor:
```suggestion
* Base aggregate function with retraction for FIRST_VALUE and LAST_VALUE.
```
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/FirstLastValueWithRetractAggFunctionBase.java
##########
@@ -0,0 +1,483 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.functions.aggfunctions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.dataview.MapView;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.binary.BinaryStringData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.BiFunction;
+
+import static org.apache.flink.table.types.utils.DataTypeUtils.toInternalDataType;
+
+/**
+ * Base retraction aggregate function for FIRST_VALUE and LAST_VALUE.
+ */
+@Internal
+abstract class FirstLastValueWithRetractAggFunctionBase<T>
+ extends InternalAggregateFunction<T, FirstLastValueWithRetractAggFunctionBase.FirstLastValueWithRetractAccumulator<T>> {
+
+ private transient DataType valueDataType;
+
+ public FirstLastValueWithRetractAggFunctionBase(LogicalType valueType) {
+ this.valueDataType = toInternalDataType(valueType);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Planning
+ // --------------------------------------------------------------------------------------------
+
+ @Override
+ public DataType[] getInputDataTypes() {
+ return new DataType[]{valueDataType};
+ }
+
+ @Override
+ public DataType getAccumulatorDataType() {
+ return DataTypes.STRUCTURED(
+ FirstLastValueWithRetractAccumulator.class,
+ DataTypes.FIELD(
+ "value",
+ valueDataType.nullable()),
+ DataTypes.FIELD(
+ "order",
+ DataTypes.BIGINT()),
+ DataTypes.FIELD(
+ "valueToOrderMap",
+ MapView.newMapViewDataType(
+ valueDataType.notNull(),
+ DataTypes.ARRAY(DataTypes.BIGINT()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "valueToOrderRetractMap",
+ MapView.newMapViewDataType(
+ valueDataType.notNull(),
+ DataTypes.ARRAY(DataTypes.BIGINT()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "orderToValueMap",
+ MapView.newMapViewDataType(
+ DataTypes.BIGINT(),
+ DataTypes.ARRAY(valueDataType.notNull()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "orderToValueRetractMap",
+ MapView.newMapViewDataType(
+ DataTypes.BIGINT(),
+ DataTypes.ARRAY(valueDataType.notNull()).bridgedTo(List.class)))
+ );
+ }
+
+ @Override
+ public DataType getOutputDataType() {
+ return valueDataType;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Runtime
+ // --------------------------------------------------------------------------------------------
+
+ /** Accumulator for FIRST_VALUE or LAST_VALUE. */
+ public static class FirstLastValueWithRetractAccumulator<T> {
+ public T value = null;
+ public Long order = null;
+ public MapView<T, List<Long>> valueToOrderMap = new MapView<>();
+ public MapView<T, List<Long>> valueToOrderRetractMap = new MapView<>();
+ public MapView<Long, List<T>> orderToValueMap = new MapView<>();
+ public MapView<Long, List<T>> orderToValueRetractMap = new MapView<>();
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof FirstLastValueWithRetractAccumulator)) {
+ return false;
+ }
+ FirstLastValueWithRetractAccumulator<?> that = (FirstLastValueWithRetractAccumulator<?>) o;
+ return Objects.equals(value, that.value) &&
+ Objects.equals(order, that.order) &&
+ Objects.equals(valueToOrderMap, that.valueToOrderMap) &&
+ Objects.equals(valueToOrderRetractMap, that.valueToOrderRetractMap) &&
+ Objects.equals(orderToValueMap, that.orderToValueMap) &&
+ Objects.equals(orderToValueRetractMap, that.orderToValueRetractMap);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ value,
+ order,
+ valueToOrderMap,
+ valueToOrderRetractMap,
+ orderToValueMap,
+ orderToValueRetractMap);
+ }
+ }
+
+ @Override
+ public FirstLastValueWithRetractAccumulator<T> createAccumulator() {
+ return new FirstLastValueWithRetractAccumulator<>();
+ }
+
+ @SuppressWarnings("unchecked")
+ public void accumulate(
+ FirstLastValueWithRetractAccumulator<T> acc,
+ Object value,
+ BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ Long order = System.nanoTime();
Review comment:
```suggestion
Long order = System.currentTimeMillis();
```
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/FirstLastValueWithRetractAggFunctionBase.java
##########
@@ -0,0 +1,483 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.functions.aggfunctions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.dataview.MapView;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.binary.BinaryStringData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.BiFunction;
+
+import static org.apache.flink.table.types.utils.DataTypeUtils.toInternalDataType;
+
+/**
+ * Base retraction aggregate function for FIRST_VALUE and LAST_VALUE.
+ */
+@Internal
+abstract class FirstLastValueWithRetractAggFunctionBase<T>
+ extends InternalAggregateFunction<T, FirstLastValueWithRetractAggFunctionBase.FirstLastValueWithRetractAccumulator<T>> {
+
+ private transient DataType valueDataType;
+
+ public FirstLastValueWithRetractAggFunctionBase(LogicalType valueType) {
+ this.valueDataType = toInternalDataType(valueType);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Planning
+ // --------------------------------------------------------------------------------------------
+
+ @Override
+ public DataType[] getInputDataTypes() {
+ return new DataType[]{valueDataType};
+ }
+
+ @Override
+ public DataType getAccumulatorDataType() {
+ return DataTypes.STRUCTURED(
+ FirstLastValueWithRetractAccumulator.class,
+ DataTypes.FIELD(
+ "value",
+ valueDataType.nullable()),
+ DataTypes.FIELD(
+ "order",
+ DataTypes.BIGINT()),
+ DataTypes.FIELD(
+ "valueToOrderMap",
+ MapView.newMapViewDataType(
+ valueDataType.notNull(),
+ DataTypes.ARRAY(DataTypes.BIGINT()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "valueToOrderRetractMap",
+ MapView.newMapViewDataType(
+ valueDataType.notNull(),
+ DataTypes.ARRAY(DataTypes.BIGINT()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "orderToValueMap",
+ MapView.newMapViewDataType(
+ DataTypes.BIGINT(),
+ DataTypes.ARRAY(valueDataType.notNull()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "orderToValueRetractMap",
+ MapView.newMapViewDataType(
+ DataTypes.BIGINT(),
+ DataTypes.ARRAY(valueDataType.notNull()).bridgedTo(List.class)))
+ );
+ }
+
+ @Override
+ public DataType getOutputDataType() {
+ return valueDataType;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Runtime
+ // --------------------------------------------------------------------------------------------
+
+ /** Accumulator for FIRST_VALUE or LAST_VALUE. */
+ public static class FirstLastValueWithRetractAccumulator<T> {
+ public T value = null;
+ public Long order = null;
+ public MapView<T, List<Long>> valueToOrderMap = new MapView<>();
+ public MapView<T, List<Long>> valueToOrderRetractMap = new MapView<>();
+ public MapView<Long, List<T>> orderToValueMap = new MapView<>();
+ public MapView<Long, List<T>> orderToValueRetractMap = new MapView<>();
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof FirstLastValueWithRetractAccumulator)) {
+ return false;
+ }
+ FirstLastValueWithRetractAccumulator<?> that = (FirstLastValueWithRetractAccumulator<?>) o;
+ return Objects.equals(value, that.value) &&
+ Objects.equals(order, that.order) &&
+ Objects.equals(valueToOrderMap, that.valueToOrderMap) &&
+ Objects.equals(valueToOrderRetractMap, that.valueToOrderRetractMap) &&
+ Objects.equals(orderToValueMap, that.orderToValueMap) &&
+ Objects.equals(orderToValueRetractMap, that.orderToValueRetractMap);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ value,
+ order,
+ valueToOrderMap,
+ valueToOrderRetractMap,
+ orderToValueMap,
+ orderToValueRetractMap);
+ }
+ }
+
+ @Override
+ public FirstLastValueWithRetractAccumulator<T> createAccumulator() {
+ return new FirstLastValueWithRetractAccumulator<>();
+ }
+
+ @SuppressWarnings("unchecked")
+ public void accumulate(
+ FirstLastValueWithRetractAccumulator<T> acc,
+ Object value,
+ BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ Long order = System.nanoTime();
+ accumulate(acc, v, order, compare);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void accumulate(
+ FirstLastValueWithRetractAccumulator<T> acc,
+ Object value, Long order,
+ BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ Long prevOrder = acc.order;
+ if (prevOrder == null || compare.apply(prevOrder, order)) {
+ acc.value = v;
+ acc.order = order;
+ }
+
+ // save order to orderToValueMap
+ List<Long> orderList = acc.valueToOrderMap.get(v);
+ if (orderList == null) {
+ orderList = new ArrayList<>();
+ orderList.add(order);
+ } else {
+ // use insertion sort,
+ // for the O(1) time complexity if order is asc, compare from back to front
+ int i = orderList.size();
+ boolean isInsert = false;
+ while (i > 0) {
+ Long tempOrder = orderList.get(i - 1);
+ if (tempOrder <= order) {
+ orderList.add(i, order);
+ isInsert = true;
+ break;
+ }
+ i--;
+ }
+ if (!isInsert) {
+ orderList.add(0, order);
+ }
+ }
+ // update valueToOrderMap
+ acc.valueToOrderMap.put(v, orderList);
+
+ // update orderToValueMap
+ List<T> valueList = acc.orderToValueMap.get(order);
+ if (valueList == null) {
+ valueList = new ArrayList<>();
+ }
+ valueList.add(v);
+ acc.orderToValueMap.put(order, valueList);
+ }
+ }
+
+ public void accumulate(FirstLastValueWithRetractAccumulator<T> acc, StringData value, BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ accumulate(acc, (Object) ((BinaryStringData) value).copy(), compare);
+ }
+ }
+
+ public void accumulate(FirstLastValueWithRetractAccumulator<T> acc, StringData value, Long order, BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ accumulate(acc, (Object) ((BinaryStringData) value).copy(), order, compare);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void retract(FirstLastValueWithRetractAccumulator<T> acc, Object value) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ List<Long> orderList = acc.valueToOrderMap.get(v);
+ if (orderList != null && orderList.size() > 0) {
+ Long order = orderList.get(0);
+ retract(acc, v, order);
+ }
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void retract(FirstLastValueWithRetractAccumulator<T> acc, Object value, Long order) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ // remove from valueToOrderMap
+ List<Long> orderList = acc.valueToOrderMap.get(v);
+ if (orderList != null && orderList.size() > 0) {
+ orderList.remove(0);
+ if (orderList.isEmpty()) {
+ acc.valueToOrderMap.remove(v);
+ } else {
+ acc.valueToOrderMap.put(v, orderList);
+ }
+ } else {
+ // if not exist, save to valueToOrderRetractMap
+ List<Long> retractOrderList = acc.valueToOrderRetractMap.get(v);
+ if (retractOrderList == null) {
+ retractOrderList = new ArrayList<>();
+ }
+ retractOrderList.add(order);
+ acc.valueToOrderRetractMap.put(v, retractOrderList);
+ }
+ // remove from orderToValueMap
+ List<T> valueList = acc.orderToValueMap.get(order);
+ int index = -1;
Review comment:
```suggestion
int index;
```
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/FirstLastValueWithRetractAggFunctionBase.java
##########
@@ -0,0 +1,483 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.functions.aggfunctions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.dataview.MapView;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.binary.BinaryStringData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.BiFunction;
+
+import static org.apache.flink.table.types.utils.DataTypeUtils.toInternalDataType;
+
+/**
+ * Base retraction aggregate function for FIRST_VALUE and LAST_VALUE.
+ */
+@Internal
+abstract class FirstLastValueWithRetractAggFunctionBase<T>
+ extends InternalAggregateFunction<T, FirstLastValueWithRetractAggFunctionBase.FirstLastValueWithRetractAccumulator<T>> {
+
+ private transient DataType valueDataType;
+
+ public FirstLastValueWithRetractAggFunctionBase(LogicalType valueType) {
+ this.valueDataType = toInternalDataType(valueType);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Planning
+ // --------------------------------------------------------------------------------------------
+
+ @Override
+ public DataType[] getInputDataTypes() {
+ return new DataType[]{valueDataType};
+ }
+
+ @Override
+ public DataType getAccumulatorDataType() {
+ return DataTypes.STRUCTURED(
+ FirstLastValueWithRetractAccumulator.class,
+ DataTypes.FIELD(
+ "value",
+ valueDataType.nullable()),
+ DataTypes.FIELD(
+ "order",
+ DataTypes.BIGINT()),
+ DataTypes.FIELD(
+ "valueToOrderMap",
+ MapView.newMapViewDataType(
+ valueDataType.notNull(),
+ DataTypes.ARRAY(DataTypes.BIGINT()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "valueToOrderRetractMap",
+ MapView.newMapViewDataType(
+ valueDataType.notNull(),
+ DataTypes.ARRAY(DataTypes.BIGINT()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "orderToValueMap",
+ MapView.newMapViewDataType(
+ DataTypes.BIGINT(),
+ DataTypes.ARRAY(valueDataType.notNull()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "orderToValueRetractMap",
+ MapView.newMapViewDataType(
+ DataTypes.BIGINT(),
+ DataTypes.ARRAY(valueDataType.notNull()).bridgedTo(List.class)))
+ );
+ }
+
+ @Override
+ public DataType getOutputDataType() {
+ return valueDataType;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Runtime
+ // --------------------------------------------------------------------------------------------
+
+ /** Accumulator for FIRST_VALUE or LAST_VALUE. */
+ public static class FirstLastValueWithRetractAccumulator<T> {
+ public T value = null;
+ public Long order = null;
+ public MapView<T, List<Long>> valueToOrderMap = new MapView<>();
+ public MapView<T, List<Long>> valueToOrderRetractMap = new MapView<>();
+ public MapView<Long, List<T>> orderToValueMap = new MapView<>();
+ public MapView<Long, List<T>> orderToValueRetractMap = new MapView<>();
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof FirstLastValueWithRetractAccumulator)) {
+ return false;
+ }
+ FirstLastValueWithRetractAccumulator<?> that = (FirstLastValueWithRetractAccumulator<?>) o;
+ return Objects.equals(value, that.value) &&
+ Objects.equals(order, that.order) &&
+ Objects.equals(valueToOrderMap, that.valueToOrderMap) &&
+ Objects.equals(valueToOrderRetractMap, that.valueToOrderRetractMap) &&
+ Objects.equals(orderToValueMap, that.orderToValueMap) &&
+ Objects.equals(orderToValueRetractMap, that.orderToValueRetractMap);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ value,
+ order,
+ valueToOrderMap,
+ valueToOrderRetractMap,
+ orderToValueMap,
+ orderToValueRetractMap);
+ }
+ }
+
+ @Override
+ public FirstLastValueWithRetractAccumulator<T> createAccumulator() {
+ return new FirstLastValueWithRetractAccumulator<>();
+ }
+
+ @SuppressWarnings("unchecked")
+ public void accumulate(
+ FirstLastValueWithRetractAccumulator<T> acc,
+ Object value,
+ BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ Long order = System.nanoTime();
+ accumulate(acc, v, order, compare);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void accumulate(
+ FirstLastValueWithRetractAccumulator<T> acc,
+ Object value, Long order,
+ BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ Long prevOrder = acc.order;
+ if (prevOrder == null || compare.apply(prevOrder, order)) {
+ acc.value = v;
+ acc.order = order;
+ }
+
+ // save order to orderToValueMap
+ List<Long> orderList = acc.valueToOrderMap.get(v);
+ if (orderList == null) {
+ orderList = new ArrayList<>();
+ orderList.add(order);
+ } else {
+ // use insertion sort,
+ // for the O(1) time complexity if order is asc, compare from back to front
+ int i = orderList.size();
+ boolean isInsert = false;
+ while (i > 0) {
+ Long tempOrder = orderList.get(i - 1);
+ if (tempOrder <= order) {
+ orderList.add(i, order);
+ isInsert = true;
+ break;
+ }
+ i--;
+ }
+ if (!isInsert) {
+ orderList.add(0, order);
+ }
+ }
+ // update valueToOrderMap
+ acc.valueToOrderMap.put(v, orderList);
+
+ // update orderToValueMap
+ List<T> valueList = acc.orderToValueMap.get(order);
+ if (valueList == null) {
+ valueList = new ArrayList<>();
+ }
+ valueList.add(v);
+ acc.orderToValueMap.put(order, valueList);
+ }
+ }
+
+ public void accumulate(FirstLastValueWithRetractAccumulator<T> acc, StringData value, BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ accumulate(acc, (Object) ((BinaryStringData) value).copy(), compare);
+ }
+ }
+
+ public void accumulate(FirstLastValueWithRetractAccumulator<T> acc, StringData value, Long order, BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ accumulate(acc, (Object) ((BinaryStringData) value).copy(), order, compare);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void retract(FirstLastValueWithRetractAccumulator<T> acc, Object value) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ List<Long> orderList = acc.valueToOrderMap.get(v);
+ if (orderList != null && orderList.size() > 0) {
+ Long order = orderList.get(0);
+ retract(acc, v, order);
+ }
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void retract(FirstLastValueWithRetractAccumulator<T> acc, Object value, Long order) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ // remove from valueToOrderMap
+ List<Long> orderList = acc.valueToOrderMap.get(v);
+ if (orderList != null && orderList.size() > 0) {
+ orderList.remove(0);
+ if (orderList.isEmpty()) {
+ acc.valueToOrderMap.remove(v);
+ } else {
+ acc.valueToOrderMap.put(v, orderList);
+ }
+ } else {
+ // if not exist, save to valueToOrderRetractMap
Review comment:
exists
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/FirstLastValueWithRetractAggFunctionBase.java
##########
@@ -0,0 +1,483 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.functions.aggfunctions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.dataview.MapView;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.binary.BinaryStringData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.BiFunction;
+
+import static org.apache.flink.table.types.utils.DataTypeUtils.toInternalDataType;
+
+/**
+ * Base retraction aggregate function for FIRST_VALUE and LAST_VALUE.
+ */
+@Internal
+abstract class FirstLastValueWithRetractAggFunctionBase<T>
+ extends InternalAggregateFunction<T, FirstLastValueWithRetractAggFunctionBase.FirstLastValueWithRetractAccumulator<T>> {
+
+ private transient DataType valueDataType;
+
+ public FirstLastValueWithRetractAggFunctionBase(LogicalType valueType) {
+ this.valueDataType = toInternalDataType(valueType);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Planning
+ // --------------------------------------------------------------------------------------------
+
+ @Override
+ public DataType[] getInputDataTypes() {
+ return new DataType[]{valueDataType};
+ }
+
+ @Override
+ public DataType getAccumulatorDataType() {
+ return DataTypes.STRUCTURED(
+ FirstLastValueWithRetractAccumulator.class,
+ DataTypes.FIELD(
+ "value",
+ valueDataType.nullable()),
+ DataTypes.FIELD(
+ "order",
+ DataTypes.BIGINT()),
+ DataTypes.FIELD(
+ "valueToOrderMap",
+ MapView.newMapViewDataType(
+ valueDataType.notNull(),
+ DataTypes.ARRAY(DataTypes.BIGINT()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "valueToOrderRetractMap",
+ MapView.newMapViewDataType(
+ valueDataType.notNull(),
+ DataTypes.ARRAY(DataTypes.BIGINT()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "orderToValueMap",
+ MapView.newMapViewDataType(
+ DataTypes.BIGINT(),
+ DataTypes.ARRAY(valueDataType.notNull()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "orderToValueRetractMap",
+ MapView.newMapViewDataType(
+ DataTypes.BIGINT(),
+ DataTypes.ARRAY(valueDataType.notNull()).bridgedTo(List.class)))
+ );
+ }
+
+ @Override
+ public DataType getOutputDataType() {
+ return valueDataType;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Runtime
+ // --------------------------------------------------------------------------------------------
+
+ /** Accumulator for FIRST_VALUE or LAST_VALUE. */
+ public static class FirstLastValueWithRetractAccumulator<T> {
+ public T value = null;
+ public Long order = null;
+ public MapView<T, List<Long>> valueToOrderMap = new MapView<>();
+ public MapView<T, List<Long>> valueToOrderRetractMap = new MapView<>();
+ public MapView<Long, List<T>> orderToValueMap = new MapView<>();
+ public MapView<Long, List<T>> orderToValueRetractMap = new MapView<>();
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof FirstLastValueWithRetractAccumulator)) {
+ return false;
+ }
+ FirstLastValueWithRetractAccumulator<?> that = (FirstLastValueWithRetractAccumulator<?>) o;
+ return Objects.equals(value, that.value) &&
+ Objects.equals(order, that.order) &&
+ Objects.equals(valueToOrderMap, that.valueToOrderMap) &&
+ Objects.equals(valueToOrderRetractMap, that.valueToOrderRetractMap) &&
+ Objects.equals(orderToValueMap, that.orderToValueMap) &&
+ Objects.equals(orderToValueRetractMap, that.orderToValueRetractMap);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ value,
+ order,
+ valueToOrderMap,
+ valueToOrderRetractMap,
+ orderToValueMap,
+ orderToValueRetractMap);
+ }
+ }
+
+ @Override
+ public FirstLastValueWithRetractAccumulator<T> createAccumulator() {
+ return new FirstLastValueWithRetractAccumulator<>();
+ }
+
+ @SuppressWarnings("unchecked")
+ public void accumulate(
+ FirstLastValueWithRetractAccumulator<T> acc,
+ Object value,
+ BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ Long order = System.nanoTime();
+ accumulate(acc, v, order, compare);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void accumulate(
+ FirstLastValueWithRetractAccumulator<T> acc,
+ Object value, Long order,
+ BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ Long prevOrder = acc.order;
+ if (prevOrder == null || compare.apply(prevOrder, order)) {
+ acc.value = v;
+ acc.order = order;
+ }
+
+ // save order to orderToValueMap
+ List<Long> orderList = acc.valueToOrderMap.get(v);
+ if (orderList == null) {
+ orderList = new ArrayList<>();
+ orderList.add(order);
+ } else {
+ // use insertion sort,
+ // for the O(1) time complexity if order is asc, compare from back to front
+ int i = orderList.size();
+ boolean isInsert = false;
+ while (i > 0) {
+ Long tempOrder = orderList.get(i - 1);
+ if (tempOrder <= order) {
+ orderList.add(i, order);
+ isInsert = true;
+ break;
+ }
+ i--;
+ }
+ if (!isInsert) {
+ orderList.add(0, order);
+ }
+ }
+ // update valueToOrderMap
+ acc.valueToOrderMap.put(v, orderList);
+
+ // update orderToValueMap
+ List<T> valueList = acc.orderToValueMap.get(order);
+ if (valueList == null) {
+ valueList = new ArrayList<>();
+ }
+ valueList.add(v);
+ acc.orderToValueMap.put(order, valueList);
+ }
+ }
+
+ public void accumulate(FirstLastValueWithRetractAccumulator<T> acc, StringData value, BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ accumulate(acc, (Object) ((BinaryStringData) value).copy(), compare);
+ }
+ }
+
+ public void accumulate(FirstLastValueWithRetractAccumulator<T> acc, StringData value, Long order, BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ accumulate(acc, (Object) ((BinaryStringData) value).copy(), order, compare);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void retract(FirstLastValueWithRetractAccumulator<T> acc, Object value) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ List<Long> orderList = acc.valueToOrderMap.get(v);
+ if (orderList != null && orderList.size() > 0) {
+ Long order = orderList.get(0);
+ retract(acc, v, order);
+ }
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void retract(FirstLastValueWithRetractAccumulator<T> acc, Object value, Long order) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ // remove from valueToOrderMap
+ List<Long> orderList = acc.valueToOrderMap.get(v);
+ if (orderList != null && orderList.size() > 0) {
+ orderList.remove(0);
+ if (orderList.isEmpty()) {
+ acc.valueToOrderMap.remove(v);
+ } else {
+ acc.valueToOrderMap.put(v, orderList);
+ }
+ } else {
+ // if not exist, save to valueToOrderRetractMap
+ List<Long> retractOrderList = acc.valueToOrderRetractMap.get(v);
+ if (retractOrderList == null) {
+ retractOrderList = new ArrayList<>();
+ }
+ retractOrderList.add(order);
+ acc.valueToOrderRetractMap.put(v, retractOrderList);
+ }
+ // remove from orderToValueMap
+ List<T> valueList = acc.orderToValueMap.get(order);
+ int index = -1;
+ if (valueList != null && (index = valueList.indexOf(v)) >= 0) {
+ valueList.remove(index);
+ if (valueList.isEmpty()) {
+ acc.orderToValueMap.remove(order);
+ } else {
+ acc.orderToValueMap.put(order, valueList);
+ }
+ } else {
+ // if not exist, save to orderToValueRetractMap
+ List<T> retractValueList = acc.orderToValueRetractMap.get(order);
+ if (retractValueList == null) {
+ retractValueList = new ArrayList<>();
+ }
+ retractValueList.add(v);
+ acc.orderToValueRetractMap.put(order, retractValueList);
+ }
+ // if retract current first value
+ if (v.equals(acc.value)) {
+ updateValue(acc);
+ }
+ }
+ }
+
+ public void merge(
+ FirstLastValueWithRetractAccumulator<T> acc,
+ Iterable<FirstLastValueWithRetractAccumulator<T>> its,
Review comment:
`its` => `accIt`
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/LastValueWithRetractAggFunction.java
##########
@@ -19,221 +19,74 @@
package org.apache.flink.table.planner.functions.aggfunctions;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.table.api.DataTypes;
-import org.apache.flink.table.api.dataview.MapView;
import org.apache.flink.table.data.StringData;
import org.apache.flink.table.data.binary.BinaryStringData;
-import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.LogicalType;
-import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
-import java.util.Objects;
-
-import static org.apache.flink.table.types.utils.DataTypeUtils.toInternalDataType;
+import java.util.function.BiFunction;
/**
* Built-in LAST_VALUE with retraction aggregate function.
*/
@Internal
public final class LastValueWithRetractAggFunction<T>
- extends InternalAggregateFunction<T, LastValueWithRetractAggFunction.LastValueWithRetractAccumulator<T>> {
-
- private transient DataType valueDataType;
+ extends FirstLastValueWithRetractAggFunctionBase<T> {
+ private static final BiFunction<Long, Long, Boolean> comparator = (x, y) -> (x - y) < 0;
Review comment:
```suggestion
private static final BiFunction<Long, Long, Boolean> comparator = (order1, order2) -> order1 < order2;
```
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/FirstValueWithRetractAggFunction.java
##########
@@ -19,225 +19,72 @@
package org.apache.flink.table.planner.functions.aggfunctions;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.table.api.DataTypes;
-import org.apache.flink.table.api.dataview.MapView;
import org.apache.flink.table.data.StringData;
import org.apache.flink.table.data.binary.BinaryStringData;
-import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.LogicalType;
-import java.util.ArrayList;
import java.util.Iterator;
-import java.util.List;
-import java.util.Objects;
-
-import static org.apache.flink.table.types.utils.DataTypeUtils.toInternalDataType;
+import java.util.function.BiFunction;
/**
* Built-in FIRST_VALUE with retraction aggregate function.
*/
@Internal
public final class FirstValueWithRetractAggFunction<T>
- extends InternalAggregateFunction<T, FirstValueWithRetractAggFunction.FirstValueWithRetractAccumulator<T>> {
-
- private transient DataType valueDataType;
+ extends FirstLastValueWithRetractAggFunctionBase<T> {
+ private static final BiFunction<Long, Long, Boolean> comparator = (x, y) -> (x - y) > 0;
public FirstValueWithRetractAggFunction(LogicalType valueType) {
- this.valueDataType = toInternalDataType(valueType);
- }
-
- // --------------------------------------------------------------------------------------------
- // Planning
- // --------------------------------------------------------------------------------------------
-
- @Override
- public DataType[] getInputDataTypes() {
- return new DataType[]{valueDataType};
- }
-
- @Override
- public DataType getAccumulatorDataType() {
- return DataTypes.STRUCTURED(
- FirstValueWithRetractAccumulator.class,
- DataTypes.FIELD(
- "firstValue",
- valueDataType.nullable()),
- DataTypes.FIELD(
- "firstOrder",
- DataTypes.BIGINT()),
- DataTypes.FIELD(
- "valueToOrderMap",
- MapView.newMapViewDataType(
- valueDataType.notNull(),
- DataTypes.ARRAY(DataTypes.BIGINT()).bridgedTo(List.class))),
- DataTypes.FIELD(
- "orderToValueMap",
- MapView.newMapViewDataType(
- DataTypes.BIGINT(),
- DataTypes.ARRAY(valueDataType.notNull()).bridgedTo(List.class)))
- );
- }
-
- @Override
- public DataType getOutputDataType() {
- return valueDataType;
- }
-
- // --------------------------------------------------------------------------------------------
- // Runtime
- // --------------------------------------------------------------------------------------------
-
- /** Accumulator for FIRST_VALUE. */
- public static class FirstValueWithRetractAccumulator<T> {
- public T firstValue;
- public Long firstOrder;
- public MapView<T, List<Long>> valueToOrderMap;
- public MapView<Long, List<T>> orderToValueMap;
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (!(o instanceof FirstValueWithRetractAccumulator)) {
- return false;
- }
- FirstValueWithRetractAccumulator<?> that = (FirstValueWithRetractAccumulator<?>) o;
- return Objects.equals(firstValue, that.firstValue) &&
- Objects.equals(firstOrder, that.firstOrder) &&
- Objects.equals(valueToOrderMap, that.valueToOrderMap) &&
- Objects.equals(orderToValueMap, that.orderToValueMap);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(firstValue, firstOrder, valueToOrderMap, orderToValueMap);
- }
- }
-
- @Override
- public FirstValueWithRetractAccumulator<T> createAccumulator() {
- final FirstValueWithRetractAccumulator<T> acc = new FirstValueWithRetractAccumulator<>();
- acc.firstValue = null;
- acc.firstOrder = null;
- acc.valueToOrderMap = new MapView<>();
- acc.orderToValueMap = new MapView<>();
- return acc;
+ super(valueType);
}
@SuppressWarnings("unchecked")
- public void accumulate(FirstValueWithRetractAccumulator<T> acc, Object value) throws Exception {
- if (value != null) {
- T v = (T) value;
- Long order = System.currentTimeMillis();
- List<Long> orderList = acc.valueToOrderMap.get(v);
- if (orderList == null) {
- orderList = new ArrayList<>();
- }
- orderList.add(order);
- acc.valueToOrderMap.put(v, orderList);
- accumulate(acc, v, order);
- }
+ public void accumulate(FirstLastValueWithRetractAccumulator<T> acc, Object value) throws Exception {
+ accumulate(acc, value, comparator);
}
@SuppressWarnings("unchecked")
- public void accumulate(FirstValueWithRetractAccumulator<T> acc, Object value, Long order) throws Exception {
- if (value != null) {
- T v = (T) value;
- Long prevOrder = acc.firstOrder;
- if (prevOrder == null || prevOrder > order) {
- acc.firstValue = v;
- acc.firstOrder = order;
- }
-
- List<T> valueList = acc.orderToValueMap.get(order);
- if (valueList == null) {
- valueList = new ArrayList<>();
- }
- valueList.add(v);
- acc.orderToValueMap.put(order, valueList);
- }
+ public void accumulate(FirstLastValueWithRetractAccumulator<T> acc, Object value, Long order) throws Exception {
+ accumulate(acc, value, order, comparator);
}
- public void accumulate(FirstValueWithRetractAccumulator<T> acc, StringData value) throws Exception {
- if (value != null) {
- accumulate(acc, (Object) ((BinaryStringData) value).copy());
- }
+ public void accumulate(FirstLastValueWithRetractAccumulator<T> acc, StringData value) throws Exception {
+ accumulate(acc, (Object) ((BinaryStringData) value).copy(), comparator);
}
- public void accumulate(FirstValueWithRetractAccumulator<T> acc, StringData value, Long order) throws Exception {
- if (value != null) {
- accumulate(acc, (Object) ((BinaryStringData) value).copy(), order);
- }
+ public void accumulate(FirstLastValueWithRetractAccumulator<T> acc, StringData value, Long order) throws Exception {
+ accumulate(acc, (Object) ((BinaryStringData) value).copy(), order, comparator);
}
- @SuppressWarnings("unchecked")
- public void retract(FirstValueWithRetractAccumulator<T> acc, Object value) throws Exception {
- if (value != null) {
- T v = (T) value;
- List<Long> orderList = acc.valueToOrderMap.get(v);
- if (orderList != null && orderList.size() > 0) {
- Long order = orderList.get(0);
- orderList.remove(0);
- if (orderList.isEmpty()) {
- acc.valueToOrderMap.remove(v);
- } else {
- acc.valueToOrderMap.put(v, orderList);
- }
- retract(acc, v, order);
- }
- }
+ public void merge(
+ FirstLastValueWithRetractAccumulator<T> acc,
+ Iterable<FirstLastValueWithRetractAccumulator<T>> its) throws Exception {
Review comment:
`its` => `accIt`
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/FirstLastValueWithRetractAggFunctionBase.java
##########
@@ -0,0 +1,483 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.functions.aggfunctions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.dataview.MapView;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.binary.BinaryStringData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.BiFunction;
+
+import static org.apache.flink.table.types.utils.DataTypeUtils.toInternalDataType;
+
+/**
+ * Base retraction aggregate function for FIRST_VALUE and LAST_VALUE.
+ */
+@Internal
+abstract class FirstLastValueWithRetractAggFunctionBase<T>
+ extends InternalAggregateFunction<T, FirstLastValueWithRetractAggFunctionBase.FirstLastValueWithRetractAccumulator<T>> {
+
+ private transient DataType valueDataType;
+
+ public FirstLastValueWithRetractAggFunctionBase(LogicalType valueType) {
+ this.valueDataType = toInternalDataType(valueType);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Planning
+ // --------------------------------------------------------------------------------------------
+
+ @Override
+ public DataType[] getInputDataTypes() {
+ return new DataType[]{valueDataType};
+ }
+
+ @Override
+ public DataType getAccumulatorDataType() {
+ return DataTypes.STRUCTURED(
+ FirstLastValueWithRetractAccumulator.class,
+ DataTypes.FIELD(
+ "value",
+ valueDataType.nullable()),
+ DataTypes.FIELD(
+ "order",
+ DataTypes.BIGINT()),
+ DataTypes.FIELD(
+ "valueToOrderMap",
+ MapView.newMapViewDataType(
+ valueDataType.notNull(),
+ DataTypes.ARRAY(DataTypes.BIGINT()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "valueToOrderRetractMap",
+ MapView.newMapViewDataType(
+ valueDataType.notNull(),
+ DataTypes.ARRAY(DataTypes.BIGINT()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "orderToValueMap",
+ MapView.newMapViewDataType(
+ DataTypes.BIGINT(),
+ DataTypes.ARRAY(valueDataType.notNull()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "orderToValueRetractMap",
+ MapView.newMapViewDataType(
+ DataTypes.BIGINT(),
+ DataTypes.ARRAY(valueDataType.notNull()).bridgedTo(List.class)))
+ );
+ }
+
+ @Override
+ public DataType getOutputDataType() {
+ return valueDataType;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Runtime
+ // --------------------------------------------------------------------------------------------
+
+ /** Accumulator for FIRST_VALUE or LAST_VALUE. */
+ public static class FirstLastValueWithRetractAccumulator<T> {
+ public T value = null;
+ public Long order = null;
+ public MapView<T, List<Long>> valueToOrderMap = new MapView<>();
+ public MapView<T, List<Long>> valueToOrderRetractMap = new MapView<>();
+ public MapView<Long, List<T>> orderToValueMap = new MapView<>();
+ public MapView<Long, List<T>> orderToValueRetractMap = new MapView<>();
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof FirstLastValueWithRetractAccumulator)) {
+ return false;
+ }
+ FirstLastValueWithRetractAccumulator<?> that = (FirstLastValueWithRetractAccumulator<?>) o;
+ return Objects.equals(value, that.value) &&
+ Objects.equals(order, that.order) &&
+ Objects.equals(valueToOrderMap, that.valueToOrderMap) &&
+ Objects.equals(valueToOrderRetractMap, that.valueToOrderRetractMap) &&
+ Objects.equals(orderToValueMap, that.orderToValueMap) &&
+ Objects.equals(orderToValueRetractMap, that.orderToValueRetractMap);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ value,
+ order,
+ valueToOrderMap,
+ valueToOrderRetractMap,
+ orderToValueMap,
+ orderToValueRetractMap);
+ }
+ }
+
+ @Override
+ public FirstLastValueWithRetractAccumulator<T> createAccumulator() {
+ return new FirstLastValueWithRetractAccumulator<>();
+ }
+
+ @SuppressWarnings("unchecked")
+ public void accumulate(
+ FirstLastValueWithRetractAccumulator<T> acc,
+ Object value,
+ BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ Long order = System.nanoTime();
+ accumulate(acc, v, order, compare);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void accumulate(
+ FirstLastValueWithRetractAccumulator<T> acc,
+ Object value, Long order,
+ BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ Long prevOrder = acc.order;
+ if (prevOrder == null || compare.apply(prevOrder, order)) {
+ acc.value = v;
+ acc.order = order;
+ }
+
+ // save order to orderToValueMap
+ List<Long> orderList = acc.valueToOrderMap.get(v);
+ if (orderList == null) {
+ orderList = new ArrayList<>();
+ orderList.add(order);
+ } else {
+ // use insertion sort,
+ // for the O(1) time complexity if order is asc, compare from back to front
+ int i = orderList.size();
+ boolean isInsert = false;
+ while (i > 0) {
+ Long tempOrder = orderList.get(i - 1);
+ if (tempOrder <= order) {
+ orderList.add(i, order);
+ isInsert = true;
+ break;
+ }
+ i--;
+ }
+ if (!isInsert) {
+ orderList.add(0, order);
+ }
+ }
+ // update valueToOrderMap
+ acc.valueToOrderMap.put(v, orderList);
+
+ // update orderToValueMap
+ List<T> valueList = acc.orderToValueMap.get(order);
+ if (valueList == null) {
+ valueList = new ArrayList<>();
+ }
+ valueList.add(v);
+ acc.orderToValueMap.put(order, valueList);
+ }
+ }
+
+ public void accumulate(FirstLastValueWithRetractAccumulator<T> acc, StringData value, BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ accumulate(acc, (Object) ((BinaryStringData) value).copy(), compare);
+ }
+ }
+
+ public void accumulate(FirstLastValueWithRetractAccumulator<T> acc, StringData value, Long order, BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ accumulate(acc, (Object) ((BinaryStringData) value).copy(), order, compare);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void retract(FirstLastValueWithRetractAccumulator<T> acc, Object value) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ List<Long> orderList = acc.valueToOrderMap.get(v);
+ if (orderList != null && orderList.size() > 0) {
+ Long order = orderList.get(0);
+ retract(acc, v, order);
+ }
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void retract(FirstLastValueWithRetractAccumulator<T> acc, Object value, Long order) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ // remove from valueToOrderMap
+ List<Long> orderList = acc.valueToOrderMap.get(v);
+ if (orderList != null && orderList.size() > 0) {
+ orderList.remove(0);
+ if (orderList.isEmpty()) {
+ acc.valueToOrderMap.remove(v);
+ } else {
+ acc.valueToOrderMap.put(v, orderList);
+ }
+ } else {
+ // if not exist, save to valueToOrderRetractMap
+ List<Long> retractOrderList = acc.valueToOrderRetractMap.get(v);
+ if (retractOrderList == null) {
+ retractOrderList = new ArrayList<>();
+ }
+ retractOrderList.add(order);
+ acc.valueToOrderRetractMap.put(v, retractOrderList);
+ }
+ // remove from orderToValueMap
+ List<T> valueList = acc.orderToValueMap.get(order);
+ int index = -1;
+ if (valueList != null && (index = valueList.indexOf(v)) >= 0) {
+ valueList.remove(index);
+ if (valueList.isEmpty()) {
+ acc.orderToValueMap.remove(order);
+ } else {
+ acc.orderToValueMap.put(order, valueList);
+ }
+ } else {
+ // if not exist, save to orderToValueRetractMap
Review comment:
exists
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/FirstValueWithRetractAggFunction.java
##########
@@ -19,225 +19,72 @@
package org.apache.flink.table.planner.functions.aggfunctions;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.table.api.DataTypes;
-import org.apache.flink.table.api.dataview.MapView;
import org.apache.flink.table.data.StringData;
import org.apache.flink.table.data.binary.BinaryStringData;
-import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.LogicalType;
-import java.util.ArrayList;
import java.util.Iterator;
-import java.util.List;
-import java.util.Objects;
-
-import static org.apache.flink.table.types.utils.DataTypeUtils.toInternalDataType;
+import java.util.function.BiFunction;
/**
* Built-in FIRST_VALUE with retraction aggregate function.
*/
@Internal
public final class FirstValueWithRetractAggFunction<T>
- extends InternalAggregateFunction<T, FirstValueWithRetractAggFunction.FirstValueWithRetractAccumulator<T>> {
-
- private transient DataType valueDataType;
+ extends FirstLastValueWithRetractAggFunctionBase<T> {
+ private static final BiFunction<Long, Long, Boolean> comparator = (x, y) -> (x - y) > 0;
Review comment:
```suggestion
private static final BiFunction<Long, Long, Boolean> comparator = (order1, order2) -> order1 > order2;
```
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/FirstLastValueWithRetractAggFunctionBase.java
##########
@@ -0,0 +1,483 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.functions.aggfunctions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.dataview.MapView;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.binary.BinaryStringData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.BiFunction;
+
+import static org.apache.flink.table.types.utils.DataTypeUtils.toInternalDataType;
+
+/**
+ * Base retraction aggregate function for FIRST_VALUE and LAST_VALUE.
+ */
+@Internal
+abstract class FirstLastValueWithRetractAggFunctionBase<T>
+ extends InternalAggregateFunction<T, FirstLastValueWithRetractAggFunctionBase.FirstLastValueWithRetractAccumulator<T>> {
+
+ private transient DataType valueDataType;
+
+ public FirstLastValueWithRetractAggFunctionBase(LogicalType valueType) {
+ this.valueDataType = toInternalDataType(valueType);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Planning
+ // --------------------------------------------------------------------------------------------
+
+ @Override
+ public DataType[] getInputDataTypes() {
+ return new DataType[]{valueDataType};
+ }
+
+ @Override
+ public DataType getAccumulatorDataType() {
+ return DataTypes.STRUCTURED(
+ FirstLastValueWithRetractAccumulator.class,
+ DataTypes.FIELD(
+ "value",
+ valueDataType.nullable()),
+ DataTypes.FIELD(
+ "order",
+ DataTypes.BIGINT()),
+ DataTypes.FIELD(
+ "valueToOrderMap",
+ MapView.newMapViewDataType(
+ valueDataType.notNull(),
+ DataTypes.ARRAY(DataTypes.BIGINT()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "valueToOrderRetractMap",
+ MapView.newMapViewDataType(
+ valueDataType.notNull(),
+ DataTypes.ARRAY(DataTypes.BIGINT()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "orderToValueMap",
+ MapView.newMapViewDataType(
+ DataTypes.BIGINT(),
+ DataTypes.ARRAY(valueDataType.notNull()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "orderToValueRetractMap",
+ MapView.newMapViewDataType(
+ DataTypes.BIGINT(),
+ DataTypes.ARRAY(valueDataType.notNull()).bridgedTo(List.class)))
+ );
+ }
+
+ @Override
+ public DataType getOutputDataType() {
+ return valueDataType;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Runtime
+ // --------------------------------------------------------------------------------------------
+
+ /** Accumulator for FIRST_VALUE or LAST_VALUE. */
+ public static class FirstLastValueWithRetractAccumulator<T> {
+ public T value = null;
+ public Long order = null;
+ public MapView<T, List<Long>> valueToOrderMap = new MapView<>();
+ public MapView<T, List<Long>> valueToOrderRetractMap = new MapView<>();
+ public MapView<Long, List<T>> orderToValueMap = new MapView<>();
+ public MapView<Long, List<T>> orderToValueRetractMap = new MapView<>();
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof FirstLastValueWithRetractAccumulator)) {
+ return false;
+ }
+ FirstLastValueWithRetractAccumulator<?> that = (FirstLastValueWithRetractAccumulator<?>) o;
+ return Objects.equals(value, that.value) &&
+ Objects.equals(order, that.order) &&
+ Objects.equals(valueToOrderMap, that.valueToOrderMap) &&
+ Objects.equals(valueToOrderRetractMap, that.valueToOrderRetractMap) &&
+ Objects.equals(orderToValueMap, that.orderToValueMap) &&
+ Objects.equals(orderToValueRetractMap, that.orderToValueRetractMap);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ value,
+ order,
+ valueToOrderMap,
+ valueToOrderRetractMap,
+ orderToValueMap,
+ orderToValueRetractMap);
+ }
+ }
+
+ @Override
+ public FirstLastValueWithRetractAccumulator<T> createAccumulator() {
+ return new FirstLastValueWithRetractAccumulator<>();
+ }
+
+ @SuppressWarnings("unchecked")
+ public void accumulate(
+ FirstLastValueWithRetractAccumulator<T> acc,
+ Object value,
+ BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ Long order = System.nanoTime();
+ accumulate(acc, v, order, compare);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void accumulate(
+ FirstLastValueWithRetractAccumulator<T> acc,
+ Object value, Long order,
+ BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ Long prevOrder = acc.order;
+ if (prevOrder == null || compare.apply(prevOrder, order)) {
+ acc.value = v;
+ acc.order = order;
+ }
+
+ // save order to orderToValueMap
+ List<Long> orderList = acc.valueToOrderMap.get(v);
+ if (orderList == null) {
+ orderList = new ArrayList<>();
+ orderList.add(order);
+ } else {
+ // use insertion sort,
+ // for the O(1) time complexity if order is asc, compare from back to front
+ int i = orderList.size();
+ boolean isInsert = false;
+ while (i > 0) {
+ Long tempOrder = orderList.get(i - 1);
+ if (tempOrder <= order) {
+ orderList.add(i, order);
+ isInsert = true;
+ break;
+ }
+ i--;
+ }
+ if (!isInsert) {
+ orderList.add(0, order);
+ }
+ }
+ // update valueToOrderMap
+ acc.valueToOrderMap.put(v, orderList);
+
+ // update orderToValueMap
+ List<T> valueList = acc.orderToValueMap.get(order);
+ if (valueList == null) {
+ valueList = new ArrayList<>();
+ }
+ valueList.add(v);
+ acc.orderToValueMap.put(order, valueList);
+ }
+ }
+
+ public void accumulate(FirstLastValueWithRetractAccumulator<T> acc, StringData value, BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ accumulate(acc, (Object) ((BinaryStringData) value).copy(), compare);
+ }
+ }
+
+ public void accumulate(FirstLastValueWithRetractAccumulator<T> acc, StringData value, Long order, BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ accumulate(acc, (Object) ((BinaryStringData) value).copy(), order, compare);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void retract(FirstLastValueWithRetractAccumulator<T> acc, Object value) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ List<Long> orderList = acc.valueToOrderMap.get(v);
+ if (orderList != null && orderList.size() > 0) {
+ Long order = orderList.get(0);
+ retract(acc, v, order);
+ }
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void retract(FirstLastValueWithRetractAccumulator<T> acc, Object value, Long order) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ // remove from valueToOrderMap
+ List<Long> orderList = acc.valueToOrderMap.get(v);
+ if (orderList != null && orderList.size() > 0) {
+ orderList.remove(0);
+ if (orderList.isEmpty()) {
+ acc.valueToOrderMap.remove(v);
+ } else {
+ acc.valueToOrderMap.put(v, orderList);
+ }
+ } else {
+ // if not exist, save to valueToOrderRetractMap
+ List<Long> retractOrderList = acc.valueToOrderRetractMap.get(v);
+ if (retractOrderList == null) {
+ retractOrderList = new ArrayList<>();
+ }
+ retractOrderList.add(order);
+ acc.valueToOrderRetractMap.put(v, retractOrderList);
+ }
+ // remove from orderToValueMap
+ List<T> valueList = acc.orderToValueMap.get(order);
+ int index = -1;
+ if (valueList != null && (index = valueList.indexOf(v)) >= 0) {
+ valueList.remove(index);
+ if (valueList.isEmpty()) {
+ acc.orderToValueMap.remove(order);
+ } else {
+ acc.orderToValueMap.put(order, valueList);
+ }
+ } else {
+ // if not exist, save to orderToValueRetractMap
+ List<T> retractValueList = acc.orderToValueRetractMap.get(order);
+ if (retractValueList == null) {
+ retractValueList = new ArrayList<>();
+ }
+ retractValueList.add(v);
+ acc.orderToValueRetractMap.put(order, retractValueList);
+ }
+ // if retract current first value
+ if (v.equals(acc.value)) {
+ updateValue(acc);
+ }
+ }
+ }
+
+ public void merge(
+ FirstLastValueWithRetractAccumulator<T> acc,
+ Iterable<FirstLastValueWithRetractAccumulator<T>> its,
+ BiFunction<Long, Long, Boolean> compare) throws Exception {
+ for (FirstLastValueWithRetractAccumulator<T> it : its) {
+ if (acc.order == null || (it.order != null && compare.apply(acc.order, it.order))) {
+ acc.value = it.value;
+ acc.order = it.order;
+ }
+ mergeValueToOrderMap(acc, it);
+ mergeOrderToValueMap(acc, it);
+ // if current first value is retract
Review comment:
```suggestion
// update the acc value if the first/last value has been retracted after merge
```
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/FirstLastValueWithRetractAggFunctionBase.java
##########
@@ -0,0 +1,483 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.functions.aggfunctions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.dataview.MapView;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.binary.BinaryStringData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.BiFunction;
+
+import static org.apache.flink.table.types.utils.DataTypeUtils.toInternalDataType;
+
+/**
+ * Base retraction aggregate function for FIRST_VALUE and LAST_VALUE.
+ */
+@Internal
+abstract class FirstLastValueWithRetractAggFunctionBase<T>
+ extends InternalAggregateFunction<T, FirstLastValueWithRetractAggFunctionBase.FirstLastValueWithRetractAccumulator<T>> {
+
+ private transient DataType valueDataType;
+
+ public FirstLastValueWithRetractAggFunctionBase(LogicalType valueType) {
+ this.valueDataType = toInternalDataType(valueType);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Planning
+ // --------------------------------------------------------------------------------------------
+
+ @Override
+ public DataType[] getInputDataTypes() {
+ return new DataType[]{valueDataType};
+ }
+
+ @Override
+ public DataType getAccumulatorDataType() {
+ return DataTypes.STRUCTURED(
+ FirstLastValueWithRetractAccumulator.class,
+ DataTypes.FIELD(
+ "value",
+ valueDataType.nullable()),
+ DataTypes.FIELD(
+ "order",
+ DataTypes.BIGINT()),
+ DataTypes.FIELD(
+ "valueToOrderMap",
+ MapView.newMapViewDataType(
+ valueDataType.notNull(),
+ DataTypes.ARRAY(DataTypes.BIGINT()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "valueToOrderRetractMap",
+ MapView.newMapViewDataType(
+ valueDataType.notNull(),
+ DataTypes.ARRAY(DataTypes.BIGINT()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "orderToValueMap",
+ MapView.newMapViewDataType(
+ DataTypes.BIGINT(),
+ DataTypes.ARRAY(valueDataType.notNull()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "orderToValueRetractMap",
+ MapView.newMapViewDataType(
+ DataTypes.BIGINT(),
+ DataTypes.ARRAY(valueDataType.notNull()).bridgedTo(List.class)))
+ );
+ }
+
+ @Override
+ public DataType getOutputDataType() {
+ return valueDataType;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Runtime
+ // --------------------------------------------------------------------------------------------
+
+ /** Accumulator for FIRST_VALUE or LAST_VALUE. */
+ public static class FirstLastValueWithRetractAccumulator<T> {
+ public T value = null;
+ public Long order = null;
+ public MapView<T, List<Long>> valueToOrderMap = new MapView<>();
+ public MapView<T, List<Long>> valueToOrderRetractMap = new MapView<>();
+ public MapView<Long, List<T>> orderToValueMap = new MapView<>();
+ public MapView<Long, List<T>> orderToValueRetractMap = new MapView<>();
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof FirstLastValueWithRetractAccumulator)) {
+ return false;
+ }
+ FirstLastValueWithRetractAccumulator<?> that = (FirstLastValueWithRetractAccumulator<?>) o;
+ return Objects.equals(value, that.value) &&
+ Objects.equals(order, that.order) &&
+ Objects.equals(valueToOrderMap, that.valueToOrderMap) &&
+ Objects.equals(valueToOrderRetractMap, that.valueToOrderRetractMap) &&
+ Objects.equals(orderToValueMap, that.orderToValueMap) &&
+ Objects.equals(orderToValueRetractMap, that.orderToValueRetractMap);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ value,
+ order,
+ valueToOrderMap,
+ valueToOrderRetractMap,
+ orderToValueMap,
+ orderToValueRetractMap);
+ }
+ }
+
+ @Override
+ public FirstLastValueWithRetractAccumulator<T> createAccumulator() {
+ return new FirstLastValueWithRetractAccumulator<>();
+ }
+
+ @SuppressWarnings("unchecked")
+ public void accumulate(
+ FirstLastValueWithRetractAccumulator<T> acc,
+ Object value,
+ BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ Long order = System.nanoTime();
+ accumulate(acc, v, order, compare);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void accumulate(
+ FirstLastValueWithRetractAccumulator<T> acc,
+ Object value, Long order,
+ BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ Long prevOrder = acc.order;
+ if (prevOrder == null || compare.apply(prevOrder, order)) {
+ acc.value = v;
+ acc.order = order;
+ }
+
+ // save order to orderToValueMap
+ List<Long> orderList = acc.valueToOrderMap.get(v);
+ if (orderList == null) {
+ orderList = new ArrayList<>();
+ orderList.add(order);
+ } else {
+ // use insertion sort,
+ // for the O(1) time complexity if order is asc, compare from back to front
+ int i = orderList.size();
+ boolean isInsert = false;
+ while (i > 0) {
+ Long tempOrder = orderList.get(i - 1);
+ if (tempOrder <= order) {
+ orderList.add(i, order);
+ isInsert = true;
+ break;
+ }
+ i--;
+ }
+ if (!isInsert) {
+ orderList.add(0, order);
+ }
+ }
+ // update valueToOrderMap
+ acc.valueToOrderMap.put(v, orderList);
+
+ // update orderToValueMap
+ List<T> valueList = acc.orderToValueMap.get(order);
+ if (valueList == null) {
+ valueList = new ArrayList<>();
+ }
+ valueList.add(v);
+ acc.orderToValueMap.put(order, valueList);
+ }
+ }
+
+ public void accumulate(FirstLastValueWithRetractAccumulator<T> acc, StringData value, BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ accumulate(acc, (Object) ((BinaryStringData) value).copy(), compare);
+ }
+ }
+
+ public void accumulate(FirstLastValueWithRetractAccumulator<T> acc, StringData value, Long order, BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ accumulate(acc, (Object) ((BinaryStringData) value).copy(), order, compare);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void retract(FirstLastValueWithRetractAccumulator<T> acc, Object value) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ List<Long> orderList = acc.valueToOrderMap.get(v);
+ if (orderList != null && orderList.size() > 0) {
+ Long order = orderList.get(0);
+ retract(acc, v, order);
+ }
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void retract(FirstLastValueWithRetractAccumulator<T> acc, Object value, Long order) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ // remove from valueToOrderMap
+ List<Long> orderList = acc.valueToOrderMap.get(v);
+ if (orderList != null && orderList.size() > 0) {
+ orderList.remove(0);
+ if (orderList.isEmpty()) {
+ acc.valueToOrderMap.remove(v);
+ } else {
+ acc.valueToOrderMap.put(v, orderList);
+ }
+ } else {
+ // if not exist, save to valueToOrderRetractMap
+ List<Long> retractOrderList = acc.valueToOrderRetractMap.get(v);
+ if (retractOrderList == null) {
+ retractOrderList = new ArrayList<>();
+ }
+ retractOrderList.add(order);
+ acc.valueToOrderRetractMap.put(v, retractOrderList);
+ }
+ // remove from orderToValueMap
+ List<T> valueList = acc.orderToValueMap.get(order);
+ int index = -1;
+ if (valueList != null && (index = valueList.indexOf(v)) >= 0) {
+ valueList.remove(index);
+ if (valueList.isEmpty()) {
+ acc.orderToValueMap.remove(order);
+ } else {
+ acc.orderToValueMap.put(order, valueList);
+ }
+ } else {
+ // if not exist, save to orderToValueRetractMap
+ List<T> retractValueList = acc.orderToValueRetractMap.get(order);
+ if (retractValueList == null) {
+ retractValueList = new ArrayList<>();
+ }
+ retractValueList.add(v);
+ acc.orderToValueRetractMap.put(order, retractValueList);
+ }
+ // if retract current first value
+ if (v.equals(acc.value)) {
+ updateValue(acc);
+ }
+ }
+ }
+
+ public void merge(
+ FirstLastValueWithRetractAccumulator<T> acc,
+ Iterable<FirstLastValueWithRetractAccumulator<T>> its,
+ BiFunction<Long, Long, Boolean> compare) throws Exception {
+ for (FirstLastValueWithRetractAccumulator<T> it : its) {
+ if (acc.order == null || (it.order != null && compare.apply(acc.order, it.order))) {
+ acc.value = it.value;
+ acc.order = it.order;
+ }
+ mergeValueToOrderMap(acc, it);
+ mergeOrderToValueMap(acc, it);
+ // if current first value is retract
+ if (!acc.valueToOrderMap.contains(acc.value)) {
+ updateValue(acc);
+ }
+ }
+ }
+
+ public void resetAccumulator(FirstLastValueWithRetractAccumulator<T> acc) {
+ acc.value = null;
+ acc.order = null;
+ acc.valueToOrderMap.clear();
+ acc.valueToOrderRetractMap.clear();
+ acc.orderToValueMap.clear();
+ acc.orderToValueRetractMap.clear();
+ }
+
+ @Override
+ public T getValue(FirstLastValueWithRetractAccumulator<T> acc) {
+ return acc.value;
+ }
+
+ private void mergeOrderToValueMap(
+ FirstLastValueWithRetractAccumulator<T> acc,
+ FirstLastValueWithRetractAccumulator<T> it) throws Exception {
Review comment:
`it ` -> `otherAcc`
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/FirstLastValueWithRetractAggFunctionBase.java
##########
@@ -0,0 +1,483 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.functions.aggfunctions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.dataview.MapView;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.binary.BinaryStringData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.BiFunction;
+
+import static org.apache.flink.table.types.utils.DataTypeUtils.toInternalDataType;
+
+/**
+ * Base retraction aggregate function for FIRST_VALUE and LAST_VALUE.
+ */
+@Internal
+abstract class FirstLastValueWithRetractAggFunctionBase<T>
+ extends InternalAggregateFunction<T, FirstLastValueWithRetractAggFunctionBase.FirstLastValueWithRetractAccumulator<T>> {
+
+ private transient DataType valueDataType;
+
+ public FirstLastValueWithRetractAggFunctionBase(LogicalType valueType) {
+ this.valueDataType = toInternalDataType(valueType);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Planning
+ // --------------------------------------------------------------------------------------------
+
+ @Override
+ public DataType[] getInputDataTypes() {
+ return new DataType[]{valueDataType};
+ }
+
+ @Override
+ public DataType getAccumulatorDataType() {
+ return DataTypes.STRUCTURED(
+ FirstLastValueWithRetractAccumulator.class,
+ DataTypes.FIELD(
+ "value",
+ valueDataType.nullable()),
+ DataTypes.FIELD(
+ "order",
+ DataTypes.BIGINT()),
+ DataTypes.FIELD(
+ "valueToOrderMap",
+ MapView.newMapViewDataType(
+ valueDataType.notNull(),
+ DataTypes.ARRAY(DataTypes.BIGINT()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "valueToOrderRetractMap",
+ MapView.newMapViewDataType(
+ valueDataType.notNull(),
+ DataTypes.ARRAY(DataTypes.BIGINT()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "orderToValueMap",
+ MapView.newMapViewDataType(
+ DataTypes.BIGINT(),
+ DataTypes.ARRAY(valueDataType.notNull()).bridgedTo(List.class))),
+ DataTypes.FIELD(
+ "orderToValueRetractMap",
+ MapView.newMapViewDataType(
+ DataTypes.BIGINT(),
+ DataTypes.ARRAY(valueDataType.notNull()).bridgedTo(List.class)))
+ );
+ }
+
+ @Override
+ public DataType getOutputDataType() {
+ return valueDataType;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Runtime
+ // --------------------------------------------------------------------------------------------
+
+ /** Accumulator for FIRST_VALUE or LAST_VALUE. */
+ public static class FirstLastValueWithRetractAccumulator<T> {
+ public T value = null;
+ public Long order = null;
+ public MapView<T, List<Long>> valueToOrderMap = new MapView<>();
+ public MapView<T, List<Long>> valueToOrderRetractMap = new MapView<>();
+ public MapView<Long, List<T>> orderToValueMap = new MapView<>();
+ public MapView<Long, List<T>> orderToValueRetractMap = new MapView<>();
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof FirstLastValueWithRetractAccumulator)) {
+ return false;
+ }
+ FirstLastValueWithRetractAccumulator<?> that = (FirstLastValueWithRetractAccumulator<?>) o;
+ return Objects.equals(value, that.value) &&
+ Objects.equals(order, that.order) &&
+ Objects.equals(valueToOrderMap, that.valueToOrderMap) &&
+ Objects.equals(valueToOrderRetractMap, that.valueToOrderRetractMap) &&
+ Objects.equals(orderToValueMap, that.orderToValueMap) &&
+ Objects.equals(orderToValueRetractMap, that.orderToValueRetractMap);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ value,
+ order,
+ valueToOrderMap,
+ valueToOrderRetractMap,
+ orderToValueMap,
+ orderToValueRetractMap);
+ }
+ }
+
+ @Override
+ public FirstLastValueWithRetractAccumulator<T> createAccumulator() {
+ return new FirstLastValueWithRetractAccumulator<>();
+ }
+
+ @SuppressWarnings("unchecked")
+ public void accumulate(
+ FirstLastValueWithRetractAccumulator<T> acc,
+ Object value,
+ BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ Long order = System.nanoTime();
+ accumulate(acc, v, order, compare);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void accumulate(
+ FirstLastValueWithRetractAccumulator<T> acc,
+ Object value, Long order,
+ BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ Long prevOrder = acc.order;
+ if (prevOrder == null || compare.apply(prevOrder, order)) {
+ acc.value = v;
+ acc.order = order;
+ }
+
+ // save order to orderToValueMap
+ List<Long> orderList = acc.valueToOrderMap.get(v);
+ if (orderList == null) {
+ orderList = new ArrayList<>();
+ orderList.add(order);
+ } else {
+ // use insertion sort,
+ // for the O(1) time complexity if order is asc, compare from back to front
+ int i = orderList.size();
+ boolean isInsert = false;
+ while (i > 0) {
+ Long tempOrder = orderList.get(i - 1);
+ if (tempOrder <= order) {
+ orderList.add(i, order);
+ isInsert = true;
+ break;
+ }
+ i--;
+ }
+ if (!isInsert) {
+ orderList.add(0, order);
+ }
+ }
+ // update valueToOrderMap
+ acc.valueToOrderMap.put(v, orderList);
+
+ // update orderToValueMap
+ List<T> valueList = acc.orderToValueMap.get(order);
+ if (valueList == null) {
+ valueList = new ArrayList<>();
+ }
+ valueList.add(v);
+ acc.orderToValueMap.put(order, valueList);
+ }
+ }
+
+ public void accumulate(FirstLastValueWithRetractAccumulator<T> acc, StringData value, BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ accumulate(acc, (Object) ((BinaryStringData) value).copy(), compare);
+ }
+ }
+
+ public void accumulate(FirstLastValueWithRetractAccumulator<T> acc, StringData value, Long order, BiFunction<Long, Long, Boolean> compare) throws Exception {
+ if (value != null) {
+ accumulate(acc, (Object) ((BinaryStringData) value).copy(), order, compare);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void retract(FirstLastValueWithRetractAccumulator<T> acc, Object value) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ List<Long> orderList = acc.valueToOrderMap.get(v);
+ if (orderList != null && orderList.size() > 0) {
+ Long order = orderList.get(0);
+ retract(acc, v, order);
+ }
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void retract(FirstLastValueWithRetractAccumulator<T> acc, Object value, Long order) throws Exception {
+ if (value != null) {
+ T v = (T) value;
+ // remove from valueToOrderMap
+ List<Long> orderList = acc.valueToOrderMap.get(v);
+ if (orderList != null && orderList.size() > 0) {
+ orderList.remove(0);
+ if (orderList.isEmpty()) {
+ acc.valueToOrderMap.remove(v);
+ } else {
+ acc.valueToOrderMap.put(v, orderList);
+ }
+ } else {
+ // if not exist, save to valueToOrderRetractMap
+ List<Long> retractOrderList = acc.valueToOrderRetractMap.get(v);
+ if (retractOrderList == null) {
+ retractOrderList = new ArrayList<>();
+ }
+ retractOrderList.add(order);
+ acc.valueToOrderRetractMap.put(v, retractOrderList);
+ }
+ // remove from orderToValueMap
+ List<T> valueList = acc.orderToValueMap.get(order);
+ int index = -1;
+ if (valueList != null && (index = valueList.indexOf(v)) >= 0) {
+ valueList.remove(index);
+ if (valueList.isEmpty()) {
+ acc.orderToValueMap.remove(order);
+ } else {
+ acc.orderToValueMap.put(order, valueList);
+ }
+ } else {
+ // if not exist, save to orderToValueRetractMap
+ List<T> retractValueList = acc.orderToValueRetractMap.get(order);
+ if (retractValueList == null) {
+ retractValueList = new ArrayList<>();
+ }
+ retractValueList.add(v);
+ acc.orderToValueRetractMap.put(order, retractValueList);
+ }
+ // if retract current first value
+ if (v.equals(acc.value)) {
+ updateValue(acc);
+ }
+ }
+ }
+
+ public void merge(
+ FirstLastValueWithRetractAccumulator<T> acc,
+ Iterable<FirstLastValueWithRetractAccumulator<T>> its,
+ BiFunction<Long, Long, Boolean> compare) throws Exception {
+ for (FirstLastValueWithRetractAccumulator<T> it : its) {
+ if (acc.order == null || (it.order != null && compare.apply(acc.order, it.order))) {
+ acc.value = it.value;
+ acc.order = it.order;
+ }
+ mergeValueToOrderMap(acc, it);
+ mergeOrderToValueMap(acc, it);
+ // if current first value is retract
+ if (!acc.valueToOrderMap.contains(acc.value)) {
+ updateValue(acc);
+ }
+ }
+ }
+
+ public void resetAccumulator(FirstLastValueWithRetractAccumulator<T> acc) {
+ acc.value = null;
+ acc.order = null;
+ acc.valueToOrderMap.clear();
+ acc.valueToOrderRetractMap.clear();
+ acc.orderToValueMap.clear();
+ acc.orderToValueRetractMap.clear();
+ }
+
+ @Override
+ public T getValue(FirstLastValueWithRetractAccumulator<T> acc) {
+ return acc.value;
+ }
+
+ private void mergeOrderToValueMap(
+ FirstLastValueWithRetractAccumulator<T> acc,
+ FirstLastValueWithRetractAccumulator<T> it) throws Exception {
+ // merge orderToValueMap
+ for (Map.Entry<Long, List<T>> entry : it.orderToValueMap.entries()) {
+ Long key = entry.getKey();
+ if (acc.orderToValueMap.contains(key)) {
+ List<T> itList = entry.getValue();
+ List<T> accList = acc.orderToValueMap.get(key);
+ // merge
+ List<T> mergedList = mergeOrderedList(itList, accList, new Comparator<T>() {
+ @Override
+ public int compare(T o1, T o2) {
+ try {
+ Long o1MinOrder = acc.valueToOrderMap.get(o1).get(0);
+ Long o2MinOrder = acc.valueToOrderMap.get(o2).get(0);
+ return (int) (o1MinOrder - o2MinOrder);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ });
+ acc.orderToValueMap.put(key, mergedList);
+ } else {
+ // if order doesn't exist, put to orderToValueMap
+ acc.orderToValueMap.put(key, entry.getValue());
+ }
+ }
+
+ // merge orderToValueRetractMap
+ for (Map.Entry<Long, List<T>> entry : it.orderToValueRetractMap.entries()) {
+ Long key = entry.getKey();
+ if (acc.orderToValueRetractMap.contains(key)) {
+ List<T> valueList = acc.orderToValueRetractMap.get(key);
+ valueList.addAll(entry.getValue());
+ acc.orderToValueRetractMap.put(key, valueList);
+ } else {
+ acc.orderToValueRetractMap.put(key, entry.getValue());
+ }
+ }
+
+ // retract from orderToValueRetractMap
+ Iterator<Map.Entry<Long, List<T>>> orderToValueRetractMapIt =
+ acc.orderToValueRetractMap.entries().iterator();
+ while (orderToValueRetractMapIt.hasNext()) {
+ Map.Entry<Long, List<T>> entry = orderToValueRetractMapIt.next();
+ Long key = entry.getKey();
+ if (acc.orderToValueMap.contains(key)) {
+ List<T> valueList = acc.orderToValueMap.get(key);
+ List<T> retractValueList = entry.getValue();
+ List<T> newRetractValueList = new ArrayList<>();
+ for (int i = 0; i < retractValueList.size(); i++) {
+ if (!valueList.remove(retractValueList.get(i))) {
+ newRetractValueList.add(retractValueList.get(i));
+ }
+ }
+ if (valueList.size() > 0) {
+ acc.orderToValueMap.put(key, valueList);
+ } else {
+ acc.orderToValueMap.remove(key);
+ }
+ if (newRetractValueList.size() > 0) {
+ acc.orderToValueRetractMap.put(key, newRetractValueList);
+ } else {
+ orderToValueRetractMapIt.remove();
+ }
+ }
+ }
+ }
+
+ private void mergeValueToOrderMap(
+ FirstLastValueWithRetractAccumulator<T> acc,
+ FirstLastValueWithRetractAccumulator<T> it) throws Exception {
Review comment:
`it` -> `otherAcc`
----------------------------------------------------------------
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 #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736629561
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10400",
"triggerID" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"triggerType" : "PUSH"
}, {
"hash" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10424",
"triggerID" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"triggerType" : "PUSH"
}, {
"hash" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10428",
"triggerID" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 57ed58c709400cabc75c5a6a4fe8587bca16dbb8 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10428)
<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 #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736629561
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10400",
"triggerID" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"triggerType" : "PUSH"
}, {
"hash" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10424",
"triggerID" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"triggerType" : "PUSH"
}, {
"hash" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10428",
"triggerID" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"triggerType" : "PUSH"
}, {
"hash" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10444",
"triggerID" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"triggerType" : "PUSH"
}, {
"hash" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11188",
"triggerID" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"triggerType" : "PUSH"
}, {
"hash" : "40095107c80dfdb13e7024ed2bb9a1760b4e41a4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11210",
"triggerID" : "40095107c80dfdb13e7024ed2bb9a1760b4e41a4",
"triggerType" : "PUSH"
}, {
"hash" : "ac3d2a524eef87974559521807ed0defdeb6507d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11220",
"triggerID" : "ac3d2a524eef87974559521807ed0defdeb6507d",
"triggerType" : "PUSH"
}, {
"hash" : "f03a31c7c7f46424db160faaf4daf5e9fb1b672e",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11457",
"triggerID" : "f03a31c7c7f46424db160faaf4daf5e9fb1b672e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f03a31c7c7f46424db160faaf4daf5e9fb1b672e Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11457)
<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 #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736629561
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10400",
"triggerID" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"triggerType" : "PUSH"
}, {
"hash" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10424",
"triggerID" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"triggerType" : "PUSH"
}, {
"hash" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10428",
"triggerID" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"triggerType" : "PUSH"
}, {
"hash" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10444",
"triggerID" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"triggerType" : "PUSH"
}, {
"hash" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11188",
"triggerID" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"triggerType" : "PUSH"
}, {
"hash" : "40095107c80dfdb13e7024ed2bb9a1760b4e41a4",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11210",
"triggerID" : "40095107c80dfdb13e7024ed2bb9a1760b4e41a4",
"triggerType" : "PUSH"
}, {
"hash" : "ac3d2a524eef87974559521807ed0defdeb6507d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "ac3d2a524eef87974559521807ed0defdeb6507d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 40095107c80dfdb13e7024ed2bb9a1760b4e41a4 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11210)
* ac3d2a524eef87974559521807ed0defdeb6507d 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 #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736629561
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10400",
"triggerID" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"triggerType" : "PUSH"
}, {
"hash" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10424",
"triggerID" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"triggerType" : "PUSH"
}, {
"hash" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10428",
"triggerID" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"triggerType" : "PUSH"
}, {
"hash" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10444",
"triggerID" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"triggerType" : "PUSH"
}, {
"hash" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11188",
"triggerID" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f4861b259fafb3036cdb7a94756c6408ddfb4c32 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11188)
<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] JingsongLi closed pull request #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
JingsongLi closed pull request #14280:
URL: https://github.com/apache/flink/pull/14280
--
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] wangxlong commented on a change in pull request #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
wangxlong commented on a change in pull request #14280:
URL: https://github.com/apache/flink/pull/14280#discussion_r547299348
##########
File path: flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/functions/aggfunctions/AggFunctionTestBase.java
##########
@@ -100,15 +100,15 @@ public void testAggregateWithMerge()
List<T> inputValues = inputValueSets.get(i);
T expected = expectedResults.get(i);
// equally split the vals sequence into two sequences
- Tuple2<List<T>, List<T>> splitValues = splitValues(inputValues);
+ Tuple2<List, List> splitValues = splitValues(inputValues);
List<T> firstValues = splitValues.f0;
List<T> secondValues = splitValues.f1;
+
// 1. verify merge with accumulate
+ ACC acc = accumulateValues(firstValues);
List<ACC> accumulators = new ArrayList<>();
accumulators.add(accumulateValues(secondValues));
- ACC acc = accumulateValues(firstValues);
-
Review comment:
I update I modified it because we need to split valueList which type is `T`, and we also need to split orderList which type is `Long`. I think it should be the same because of type erasure.
But it's really not elegant, or we can use `Object` type for list element for splitValues method？
----------------------------------------------------------------
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 #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736629561
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10400",
"triggerID" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"triggerType" : "PUSH"
}, {
"hash" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10424",
"triggerID" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"triggerType" : "PUSH"
}, {
"hash" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10428",
"triggerID" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"triggerType" : "PUSH"
}, {
"hash" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10444",
"triggerID" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"triggerType" : "PUSH"
}, {
"hash" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11188",
"triggerID" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"triggerType" : "PUSH"
}, {
"hash" : "40095107c80dfdb13e7024ed2bb9a1760b4e41a4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11210",
"triggerID" : "40095107c80dfdb13e7024ed2bb9a1760b4e41a4",
"triggerType" : "PUSH"
}, {
"hash" : "ac3d2a524eef87974559521807ed0defdeb6507d",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11220",
"triggerID" : "ac3d2a524eef87974559521807ed0defdeb6507d",
"triggerType" : "PUSH"
}, {
"hash" : "f03a31c7c7f46424db160faaf4daf5e9fb1b672e",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "f03a31c7c7f46424db160faaf4daf5e9fb1b672e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ac3d2a524eef87974559521807ed0defdeb6507d Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11220)
* f03a31c7c7f46424db160faaf4daf5e9fb1b672e 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] leonardBang commented on pull request #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
leonardBang commented on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-752319197
Could you help rebase this PR? @wangxlong The master branch merged a big format PR which caused the conflict
----------------------------------------------------------------
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 #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736629561
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10400",
"triggerID" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"triggerType" : "PUSH"
}, {
"hash" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10424",
"triggerID" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"triggerType" : "PUSH"
}, {
"hash" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10428",
"triggerID" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"triggerType" : "PUSH"
}, {
"hash" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10444",
"triggerID" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"triggerType" : "PUSH"
}, {
"hash" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11188",
"triggerID" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"triggerType" : "PUSH"
}, {
"hash" : "40095107c80dfdb13e7024ed2bb9a1760b4e41a4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11210",
"triggerID" : "40095107c80dfdb13e7024ed2bb9a1760b4e41a4",
"triggerType" : "PUSH"
}, {
"hash" : "ac3d2a524eef87974559521807ed0defdeb6507d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11220",
"triggerID" : "ac3d2a524eef87974559521807ed0defdeb6507d",
"triggerType" : "PUSH"
}, {
"hash" : "f03a31c7c7f46424db160faaf4daf5e9fb1b672e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11457",
"triggerID" : "f03a31c7c7f46424db160faaf4daf5e9fb1b672e",
"triggerType" : "PUSH"
}, {
"hash" : "671c2ee902f84118152ce4d038e548a23885e208",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11491",
"triggerID" : "671c2ee902f84118152ce4d038e548a23885e208",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 671c2ee902f84118152ce4d038e548a23885e208 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11491)
<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] wangxlong commented on a change in pull request #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
wangxlong commented on a change in pull request #14280:
URL: https://github.com/apache/flink/pull/14280#discussion_r547294770
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/FirstValueAggFunction.java
##########
@@ -83,7 +83,7 @@ public void accumulate(RowData rowData, Object value) {
GenericRowData acc = (GenericRowData) rowData;
if (value != null && acc.getLong(1) == Long.MAX_VALUE) {
acc.setField(0, value);
- acc.setField(1, System.currentTimeMillis());
+ acc.setField(1, System.nanoTime());
Review comment:
I prefer the higher precision of nanoTime, for we use this as the order when merging.
----------------------------------------------------------------
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 #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736606688
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 bdbf525b9e6435c9e6c9666c1c1ba0cf88ffff9a (Fri May 28 07:04:13 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](https://issues.apache.org/jira/browse/FLINK-20110).** According to the [code contribution guide](https://flink.apache.org/contributing/contribute-code.html), tickets need to be assigned before starting with the implementation work.
<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] leonardBang commented on pull request #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
leonardBang commented on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-757963293
> ping @leonardBang
Sorry for late reply, @wangxlong , the PR looks in good scope, but the `nanotime` usage should revert from my side
----------------------------------------------------------------
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 #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736629561
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b2173ba7a73dc7d79ba4f0679b7076400815d116 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 #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736629561
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10400",
"triggerID" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"triggerType" : "PUSH"
}, {
"hash" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10424",
"triggerID" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 6a0849e61034e34186780ce45dcf9157f65c0dfd Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10424)
<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 #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736606688
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 b2173ba7a73dc7d79ba4f0679b7076400815d116 (Tue Dec 01 14:58:27 UTC 2020)
**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 #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736629561
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10400",
"triggerID" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b2173ba7a73dc7d79ba4f0679b7076400815d116 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10400)
<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 #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736629561
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10400",
"triggerID" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"triggerType" : "PUSH"
}, {
"hash" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10424",
"triggerID" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"triggerType" : "PUSH"
}, {
"hash" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10428",
"triggerID" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"triggerType" : "PUSH"
}, {
"hash" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10444",
"triggerID" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"triggerType" : "PUSH"
}, {
"hash" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11188",
"triggerID" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"triggerType" : "PUSH"
}, {
"hash" : "40095107c80dfdb13e7024ed2bb9a1760b4e41a4",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "40095107c80dfdb13e7024ed2bb9a1760b4e41a4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f4861b259fafb3036cdb7a94756c6408ddfb4c32 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11188)
* 40095107c80dfdb13e7024ed2bb9a1760b4e41a4 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 #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736629561
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10400",
"triggerID" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"triggerType" : "PUSH"
}, {
"hash" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b2173ba7a73dc7d79ba4f0679b7076400815d116 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10400)
* 6a0849e61034e34186780ce45dcf9157f65c0dfd 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] JingsongLi commented on pull request #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-858256816
Hi @wangxlong , You haven't responded for a long time. Do you want to continue?
Or @leonardBang Can you continue to finish this PR?
--
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 #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736629561
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10400",
"triggerID" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"triggerType" : "PUSH"
}, {
"hash" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10424",
"triggerID" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"triggerType" : "PUSH"
}, {
"hash" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10428",
"triggerID" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"triggerType" : "PUSH"
}, {
"hash" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10444",
"triggerID" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"triggerType" : "PUSH"
}, {
"hash" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11188",
"triggerID" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"triggerType" : "PUSH"
}, {
"hash" : "40095107c80dfdb13e7024ed2bb9a1760b4e41a4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11210",
"triggerID" : "40095107c80dfdb13e7024ed2bb9a1760b4e41a4",
"triggerType" : "PUSH"
}, {
"hash" : "ac3d2a524eef87974559521807ed0defdeb6507d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11220",
"triggerID" : "ac3d2a524eef87974559521807ed0defdeb6507d",
"triggerType" : "PUSH"
}, {
"hash" : "f03a31c7c7f46424db160faaf4daf5e9fb1b672e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11457",
"triggerID" : "f03a31c7c7f46424db160faaf4daf5e9fb1b672e",
"triggerType" : "PUSH"
}, {
"hash" : "671c2ee902f84118152ce4d038e548a23885e208",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11491",
"triggerID" : "671c2ee902f84118152ce4d038e548a23885e208",
"triggerType" : "PUSH"
}, {
"hash" : "bdbf525b9e6435c9e6c9666c1c1ba0cf88ffff9a",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11496",
"triggerID" : "bdbf525b9e6435c9e6c9666c1c1ba0cf88ffff9a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* bdbf525b9e6435c9e6c9666c1c1ba0cf88ffff9a Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11496)
<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] leonardBang edited a comment on pull request #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
leonardBang edited a comment on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-757963293
> ping @leonardBang
Sorry for late reply, @wangxlong , the PR looks in good scope,
but (1) the `nanotime` usage should revert from my side (2) test `AggregateITCase.testNestedGroupByAgg` is unstable(you can run multiple times) we need make it clear
----------------------------------------------------------------
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 #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736629561
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10400",
"triggerID" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"triggerType" : "PUSH"
}, {
"hash" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10424",
"triggerID" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"triggerType" : "PUSH"
}, {
"hash" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10428",
"triggerID" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"triggerType" : "PUSH"
}, {
"hash" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10444",
"triggerID" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"triggerType" : "PUSH"
}, {
"hash" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11188",
"triggerID" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"triggerType" : "PUSH"
}, {
"hash" : "40095107c80dfdb13e7024ed2bb9a1760b4e41a4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11210",
"triggerID" : "40095107c80dfdb13e7024ed2bb9a1760b4e41a4",
"triggerType" : "PUSH"
}, {
"hash" : "ac3d2a524eef87974559521807ed0defdeb6507d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11220",
"triggerID" : "ac3d2a524eef87974559521807ed0defdeb6507d",
"triggerType" : "PUSH"
}, {
"hash" : "f03a31c7c7f46424db160faaf4daf5e9fb1b672e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11457",
"triggerID" : "f03a31c7c7f46424db160faaf4daf5e9fb1b672e",
"triggerType" : "PUSH"
}, {
"hash" : "671c2ee902f84118152ce4d038e548a23885e208",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11491",
"triggerID" : "671c2ee902f84118152ce4d038e548a23885e208",
"triggerType" : "PUSH"
}, {
"hash" : "bdbf525b9e6435c9e6c9666c1c1ba0cf88ffff9a",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11496",
"triggerID" : "bdbf525b9e6435c9e6c9666c1c1ba0cf88ffff9a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 671c2ee902f84118152ce4d038e548a23885e208 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11491)
* bdbf525b9e6435c9e6c9666c1c1ba0cf88ffff9a Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11496)
<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] wangxlong commented on a change in pull request #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
wangxlong commented on a change in pull request #14280:
URL: https://github.com/apache/flink/pull/14280#discussion_r547295921
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/FirstValueAggFunction.java
##########
@@ -107,6 +107,27 @@ public void accumulate(RowData rowData, StringData value, Long order) {
}
}
+ public void merge(RowData acc, Iterable<RowData> its) {
+ GenericRowData accRowData = (GenericRowData) acc;
+ long accOrder = accRowData.getLong(1);
+ GenericRowData firstAcc = accRowData;
+ boolean needUpdate = false;
+
+ for (RowData rowData : its) {
Review comment:
For it's not using the remove method, so It's ok to use that?
----------------------------------------------------------------
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 #14280: [FLINK-20110][table-planner] Support 'merge' method for first_value a…
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14280:
URL: https://github.com/apache/flink/pull/14280#issuecomment-736629561
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10400",
"triggerID" : "b2173ba7a73dc7d79ba4f0679b7076400815d116",
"triggerType" : "PUSH"
}, {
"hash" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10424",
"triggerID" : "6a0849e61034e34186780ce45dcf9157f65c0dfd",
"triggerType" : "PUSH"
}, {
"hash" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10428",
"triggerID" : "57ed58c709400cabc75c5a6a4fe8587bca16dbb8",
"triggerType" : "PUSH"
}, {
"hash" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10444",
"triggerID" : "851970c1c94f8f44058f914d3dbeb959cb66ee53",
"triggerType" : "PUSH"
}, {
"hash" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "f4861b259fafb3036cdb7a94756c6408ddfb4c32",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 851970c1c94f8f44058f914d3dbeb959cb66ee53 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10444)
* f4861b259fafb3036cdb7a94756c6408ddfb4c32 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