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/09/29 12:00:03 UTC
[GitHub] [flink] WeiZhong94 opened a new pull request #13507: [FLINK-19231][python] Support ListState and ListView for Python UDAF.
WeiZhong94 opened a new pull request #13507:
URL: https://github.com/apache/flink/pull/13507
## What is the purpose of the change
*This pull request supports ListState and ListView for Python UDAF*
## Brief change log
- *Support ListState and ListView for Python UDAF*
## Verifying this change
This change is already covered by existing tests*.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes)
- The serializers: (no)
- The runtime per-record code paths (performance sensitive): (yes)
- 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? (yes)
- If yes, how is the feature documented? (docs)
----------------------------------------------------------------
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] dianfu closed pull request #13507: [FLINK-19231][python] Support ListState and ListView for Python UDAF.
Posted by GitBox <gi...@apache.org>.
dianfu closed pull request #13507:
URL: https://github.com/apache/flink/pull/13507
----------------------------------------------------------------
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 #13507: [FLINK-19231][python] Support ListState and ListView for Python UDAF.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13507:
URL: https://github.com/apache/flink/pull/13507#issuecomment-700666447
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4a0ccb37d548b600efc2d5eb5d2e786f11e6bfa2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7078",
"triggerID" : "4a0ccb37d548b600efc2d5eb5d2e786f11e6bfa2",
"triggerType" : "PUSH"
}, {
"hash" : "de6024d84022ada3b5bbdbf4c6e82a47c8459d98",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7412",
"triggerID" : "de6024d84022ada3b5bbdbf4c6e82a47c8459d98",
"triggerType" : "PUSH"
}, {
"hash" : "32df357f8334410a819301bb1a1425681185a48e",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "32df357f8334410a819301bb1a1425681185a48e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* de6024d84022ada3b5bbdbf4c6e82a47c8459d98 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7412)
* 32df357f8334410a819301bb1a1425681185a48e 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 #13507: [FLINK-19231][python] Support ListState and ListView for Python UDAF.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13507:
URL: https://github.com/apache/flink/pull/13507#issuecomment-700666447
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4a0ccb37d548b600efc2d5eb5d2e786f11e6bfa2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7078",
"triggerID" : "4a0ccb37d548b600efc2d5eb5d2e786f11e6bfa2",
"triggerType" : "PUSH"
}, {
"hash" : "de6024d84022ada3b5bbdbf4c6e82a47c8459d98",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7412",
"triggerID" : "de6024d84022ada3b5bbdbf4c6e82a47c8459d98",
"triggerType" : "PUSH"
}, {
"hash" : "32df357f8334410a819301bb1a1425681185a48e",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7481",
"triggerID" : "32df357f8334410a819301bb1a1425681185a48e",
"triggerType" : "PUSH"
}, {
"hash" : "958a19ec59344cf108d39153cfe25f849fed15af",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7494",
"triggerID" : "958a19ec59344cf108d39153cfe25f849fed15af",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 32df357f8334410a819301bb1a1425681185a48e Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7481)
* 958a19ec59344cf108d39153cfe25f849fed15af Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7494)
<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 #13507: [FLINK-19231][python] Support ListState and ListView for Python UDAF.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13507:
URL: https://github.com/apache/flink/pull/13507#issuecomment-700666447
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4a0ccb37d548b600efc2d5eb5d2e786f11e6bfa2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7078",
"triggerID" : "4a0ccb37d548b600efc2d5eb5d2e786f11e6bfa2",
"triggerType" : "PUSH"
}, {
"hash" : "de6024d84022ada3b5bbdbf4c6e82a47c8459d98",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7412",
"triggerID" : "de6024d84022ada3b5bbdbf4c6e82a47c8459d98",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* de6024d84022ada3b5bbdbf4c6e82a47c8459d98 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7412)
<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] twalthr commented on pull request #13507: [FLINK-19231][python] Support ListState and ListView for Python UDAF.
Posted by GitBox <gi...@apache.org>.
twalthr commented on pull request #13507:
URL: https://github.com/apache/flink/pull/13507#issuecomment-706956988
Thanks for the update @WeiZhong94. My comment is addressed. I will leave the full review to somebody else.
----------------------------------------------------------------
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] WeiZhong94 commented on pull request #13507: [FLINK-19231][python] Support ListState and ListView for Python UDAF.
Posted by GitBox <gi...@apache.org>.
WeiZhong94 commented on pull request #13507:
URL: https://github.com/apache/flink/pull/13507#issuecomment-707457002
@dianfu Thanks for your comments. I have updated 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 #13507: [FLINK-19231][python] Support ListState and ListView for Python UDAF.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13507:
URL: https://github.com/apache/flink/pull/13507#issuecomment-700666447
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4a0ccb37d548b600efc2d5eb5d2e786f11e6bfa2",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7078",
"triggerID" : "4a0ccb37d548b600efc2d5eb5d2e786f11e6bfa2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4a0ccb37d548b600efc2d5eb5d2e786f11e6bfa2 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7078)
<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 #13507: [FLINK-19231][python] Support ListState and ListView for Python UDAF.
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #13507:
URL: https://github.com/apache/flink/pull/13507#issuecomment-700656265
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 4a0ccb37d548b600efc2d5eb5d2e786f11e6bfa2 (Tue Sep 29 12:01:50 UTC 2020)
**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-19231).** 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] flinkbot edited a comment on pull request #13507: [FLINK-19231][python] Support ListState and ListView for Python UDAF.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13507:
URL: https://github.com/apache/flink/pull/13507#issuecomment-700666447
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4a0ccb37d548b600efc2d5eb5d2e786f11e6bfa2",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7078",
"triggerID" : "4a0ccb37d548b600efc2d5eb5d2e786f11e6bfa2",
"triggerType" : "PUSH"
}, {
"hash" : "de6024d84022ada3b5bbdbf4c6e82a47c8459d98",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "de6024d84022ada3b5bbdbf4c6e82a47c8459d98",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4a0ccb37d548b600efc2d5eb5d2e786f11e6bfa2 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7078)
* de6024d84022ada3b5bbdbf4c6e82a47c8459d98 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 #13507: [FLINK-19231][python] Support ListState and ListView for Python UDAF.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13507:
URL: https://github.com/apache/flink/pull/13507#issuecomment-700666447
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4a0ccb37d548b600efc2d5eb5d2e786f11e6bfa2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7078",
"triggerID" : "4a0ccb37d548b600efc2d5eb5d2e786f11e6bfa2",
"triggerType" : "PUSH"
}, {
"hash" : "de6024d84022ada3b5bbdbf4c6e82a47c8459d98",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7412",
"triggerID" : "de6024d84022ada3b5bbdbf4c6e82a47c8459d98",
"triggerType" : "PUSH"
}, {
"hash" : "32df357f8334410a819301bb1a1425681185a48e",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7481",
"triggerID" : "32df357f8334410a819301bb1a1425681185a48e",
"triggerType" : "PUSH"
}, {
"hash" : "958a19ec59344cf108d39153cfe25f849fed15af",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7494",
"triggerID" : "958a19ec59344cf108d39153cfe25f849fed15af",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* de6024d84022ada3b5bbdbf4c6e82a47c8459d98 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7412)
* 32df357f8334410a819301bb1a1425681185a48e Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7481)
* 958a19ec59344cf108d39153cfe25f849fed15af Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7494)
<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 #13507: [FLINK-19231][python] Support ListState and ListView for Python UDAF.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13507:
URL: https://github.com/apache/flink/pull/13507#issuecomment-700666447
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4a0ccb37d548b600efc2d5eb5d2e786f11e6bfa2",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7078",
"triggerID" : "4a0ccb37d548b600efc2d5eb5d2e786f11e6bfa2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4a0ccb37d548b600efc2d5eb5d2e786f11e6bfa2 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7078)
<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 #13507: [FLINK-19231][python] Support ListState and ListView for Python UDAF.
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #13507:
URL: https://github.com/apache/flink/pull/13507#issuecomment-700666447
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4a0ccb37d548b600efc2d5eb5d2e786f11e6bfa2",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4a0ccb37d548b600efc2d5eb5d2e786f11e6bfa2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4a0ccb37d548b600efc2d5eb5d2e786f11e6bfa2 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 #13507: [FLINK-19231][python] Support ListState and ListView for Python UDAF.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13507:
URL: https://github.com/apache/flink/pull/13507#issuecomment-700666447
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4a0ccb37d548b600efc2d5eb5d2e786f11e6bfa2",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7078",
"triggerID" : "4a0ccb37d548b600efc2d5eb5d2e786f11e6bfa2",
"triggerType" : "PUSH"
}, {
"hash" : "de6024d84022ada3b5bbdbf4c6e82a47c8459d98",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7412",
"triggerID" : "de6024d84022ada3b5bbdbf4c6e82a47c8459d98",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4a0ccb37d548b600efc2d5eb5d2e786f11e6bfa2 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7078)
* de6024d84022ada3b5bbdbf4c6e82a47c8459d98 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7412)
<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] dianfu commented on a change in pull request #13507: [FLINK-19231][python] Support ListState and ListView for Python UDAF.
Posted by GitBox <gi...@apache.org>.
dianfu commented on a change in pull request #13507:
URL: https://github.com/apache/flink/pull/13507#discussion_r503156160
##########
File path: flink-python/pyflink/fn_execution/aggregate.py
##########
@@ -16,13 +16,16 @@
# limitations under the License.
################################################################################
from abc import ABC, abstractmethod
-from typing import List
+from typing import List, Dict
Review comment:
Unused import
##########
File path: flink-python/pyflink/fn_execution/aggregate.py
##########
@@ -16,13 +16,16 @@
# limitations under the License.
################################################################################
from abc import ABC, abstractmethod
-from typing import List
+from typing import List, Dict
-from apache_beam.coders import PickleCoder
+from apache_beam.coders import PickleCoder, Coder
from pyflink.common import Row, RowKind
+from pyflink.common.state import ListState
+from pyflink.fn_execution.coders import from_proto
from pyflink.fn_execution.state_impl import RemoteKeyedStateBackend
from pyflink.table import AggregateFunction, FunctionContext
+from pyflink.table.data_view import ListView, DataView
Review comment:
Unused import
##########
File path: flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/common/CommonPythonAggregate.scala
##########
@@ -65,18 +69,77 @@ trait CommonPythonAggregate extends CommonPythonBase {
* For streaming execution we extract the PythonFunctionInfo from AggregateInfo.
*/
protected def extractPythonAggregateFunctionInfosFromAggregateInfo(
- pythonAggregateInfo: AggregateInfo): PythonFunctionInfo = {
+ aggIndex: Int,
+ pythonAggregateInfo: AggregateInfo): (PythonFunctionInfo, Array[DataViewSpec]) = {
pythonAggregateInfo.function match {
case function: PythonFunction =>
- new PythonFunctionInfo(
- function,
- pythonAggregateInfo.argIndexes.map(_.asInstanceOf[AnyRef]))
+ (
+ new PythonFunctionInfo(
+ function,
+ pythonAggregateInfo.argIndexes.map(_.asInstanceOf[AnyRef])),
+ extractDataViewSpecs(
+ aggIndex,
+ function.asInstanceOf[PythonAggregateFunction].getTypeInference(null)
+ .getAccumulatorTypeStrategy.get()
+ .inferType(null).get())
+ )
case _: Count1AggFunction =>
// The count star will be treated specially in Python UDF worker
- PythonFunctionInfo.DUMMY_PLACEHOLDER
+ (PythonFunctionInfo.DUMMY_PLACEHOLDER, Array())
case _ =>
throw new TableException(
"Unsupported python aggregate function: " + pythonAggregateInfo.function)
}
}
+
+ protected def extractDataViewSpecs(
+ index: Int,
+ accType: DataType): Array[DataViewSpec] = {
+ if (!accType.isInstanceOf[FieldsDataType]) {
+ return Array()
+ }
+
+ val compositeAccType = accType.asInstanceOf[FieldsDataType]
+
+ def includesDataView(fdt: FieldsDataType): Boolean = {
+ (0 until fdt.getChildren.size()).exists(i =>
+ fdt.getChildren.get(i).getLogicalType match {
+ case row: RowType =>
Review comment:
```suggestion
case _: RowType =>
```
##########
File path: flink-python/pyflink/fn_execution/beam/beam_operations_slow.py
##########
@@ -304,9 +306,6 @@ def finish(self):
def reset(self):
super().reset()
- if self.keyed_state_backend:
Review comment:
OK. We could just remove the reset method as it does nothing right now.
##########
File path: flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/common/CommonPythonAggregate.scala
##########
@@ -65,18 +69,77 @@ trait CommonPythonAggregate extends CommonPythonBase {
* For streaming execution we extract the PythonFunctionInfo from AggregateInfo.
*/
protected def extractPythonAggregateFunctionInfosFromAggregateInfo(
- pythonAggregateInfo: AggregateInfo): PythonFunctionInfo = {
+ aggIndex: Int,
+ pythonAggregateInfo: AggregateInfo): (PythonFunctionInfo, Array[DataViewSpec]) = {
pythonAggregateInfo.function match {
case function: PythonFunction =>
- new PythonFunctionInfo(
- function,
- pythonAggregateInfo.argIndexes.map(_.asInstanceOf[AnyRef]))
+ (
+ new PythonFunctionInfo(
+ function,
+ pythonAggregateInfo.argIndexes.map(_.asInstanceOf[AnyRef])),
+ extractDataViewSpecs(
+ aggIndex,
+ function.asInstanceOf[PythonAggregateFunction].getTypeInference(null)
+ .getAccumulatorTypeStrategy.get()
+ .inferType(null).get())
+ )
case _: Count1AggFunction =>
// The count star will be treated specially in Python UDF worker
- PythonFunctionInfo.DUMMY_PLACEHOLDER
+ (PythonFunctionInfo.DUMMY_PLACEHOLDER, Array())
case _ =>
throw new TableException(
"Unsupported python aggregate function: " + pythonAggregateInfo.function)
}
}
+
+ protected def extractDataViewSpecs(
+ index: Int,
+ accType: DataType): Array[DataViewSpec] = {
+ if (!accType.isInstanceOf[FieldsDataType]) {
+ return Array()
+ }
+
+ val compositeAccType = accType.asInstanceOf[FieldsDataType]
+
+ def includesDataView(fdt: FieldsDataType): Boolean = {
+ (0 until fdt.getChildren.size()).exists(i =>
+ fdt.getChildren.get(i).getLogicalType match {
+ case row: RowType =>
+ includesDataView(fdt.getChildren.get(i).asInstanceOf[FieldsDataType])
+ case structed: StructuredType =>
Review comment:
```suggestion
case structuredType: StructuredType =>
```
##########
File path: flink-python/pyflink/fn_execution/aggregate.py
##########
@@ -34,6 +37,75 @@ def join_row(left: Row, right: Row):
return Row(*fields)
+def extract_data_view_specs(udfs):
+ extracted_udf_data_view_specs = []
+ for udf in udfs:
+ udf_data_view_specs_proto = udf.specs
+ if udf_data_view_specs_proto is None:
+ extracted_udf_data_view_specs.append([])
+ extracted_specs = []
+ for spec_proto in udf_data_view_specs_proto:
+ state_id = spec_proto.name
+ field_index = spec_proto.field_index
+ if spec_proto.list_view is not None:
+ element_coder = from_proto(spec_proto.list_view.element_type)
+ extracted_specs.append(ListViewSpec(state_id, field_index, element_coder))
+ elif spec_proto.map_view is not None:
+ key_coder = from_proto(spec_proto.map_view.key_type)
+ value_coder = from_proto(spec_proto.map_view.value_type)
+ extracted_specs.append(MapViewSpec(state_id, field_index, key_coder, value_coder))
+ else:
+ raise Exception("Unsupported data view spec type: " + spec_proto.type)
+ extracted_udf_data_view_specs.append(extracted_specs)
+ if all([len(i) == 0 for i in extracted_udf_data_view_specs]):
+ return []
+ return extracted_udf_data_view_specs
+
+
+class StateListView(ListView):
+
+ def __init__(self, list_state):
+ super().__init__()
+ self._list_state = list_state # type: ListState
Review comment:
Why not use typehint directly?
##########
File path: flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/common/CommonPythonAggregate.scala
##########
@@ -65,18 +69,77 @@ trait CommonPythonAggregate extends CommonPythonBase {
* For streaming execution we extract the PythonFunctionInfo from AggregateInfo.
*/
protected def extractPythonAggregateFunctionInfosFromAggregateInfo(
- pythonAggregateInfo: AggregateInfo): PythonFunctionInfo = {
+ aggIndex: Int,
+ pythonAggregateInfo: AggregateInfo): (PythonFunctionInfo, Array[DataViewSpec]) = {
pythonAggregateInfo.function match {
case function: PythonFunction =>
- new PythonFunctionInfo(
- function,
- pythonAggregateInfo.argIndexes.map(_.asInstanceOf[AnyRef]))
+ (
+ new PythonFunctionInfo(
+ function,
+ pythonAggregateInfo.argIndexes.map(_.asInstanceOf[AnyRef])),
+ extractDataViewSpecs(
+ aggIndex,
+ function.asInstanceOf[PythonAggregateFunction].getTypeInference(null)
+ .getAccumulatorTypeStrategy.get()
+ .inferType(null).get())
+ )
case _: Count1AggFunction =>
// The count star will be treated specially in Python UDF worker
- PythonFunctionInfo.DUMMY_PLACEHOLDER
+ (PythonFunctionInfo.DUMMY_PLACEHOLDER, Array())
case _ =>
throw new TableException(
"Unsupported python aggregate function: " + pythonAggregateInfo.function)
}
}
+
+ protected def extractDataViewSpecs(
+ index: Int,
+ accType: DataType): Array[DataViewSpec] = {
+ if (!accType.isInstanceOf[FieldsDataType]) {
+ return Array()
+ }
+
+ val compositeAccType = accType.asInstanceOf[FieldsDataType]
+
+ def includesDataView(fdt: FieldsDataType): Boolean = {
+ (0 until fdt.getChildren.size()).exists(i =>
+ fdt.getChildren.get(i).getLogicalType match {
+ case row: RowType =>
+ includesDataView(fdt.getChildren.get(i).asInstanceOf[FieldsDataType])
+ case structed: StructuredType =>
+ classOf[DataView].isAssignableFrom(structed.getImplementationClass.get())
+ case _ => false
+ }
+ )
+ }
+
+ if (includesDataView(compositeAccType)) {
+ compositeAccType.getLogicalType match {
+ case rowType: RowType =>
+ (0 until compositeAccType.getChildren.size()).flatMap(i => {
+ compositeAccType.getChildren.get(i).getLogicalType match {
+ case _: RowType if includesDataView(
+ compositeAccType.getChildren.get(i).asInstanceOf[FieldsDataType]) =>
+ throw new TableException(
+ "For Python AggregateFunction DataView only supported at first " +
Review comment:
```suggestion
"For Python AggregateFunction, DataView cannot be used in the nested columns of the accumulator." +
```
----------------------------------------------------------------
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] twalthr commented on a change in pull request #13507: [FLINK-19231][python] Support ListState and ListView for Python UDAF.
Posted by GitBox <gi...@apache.org>.
twalthr commented on a change in pull request #13507:
URL: https://github.com/apache/flink/pull/13507#discussion_r497238937
##########
File path: flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/common/CommonPythonAggregate.scala
##########
@@ -65,18 +72,69 @@ trait CommonPythonAggregate extends CommonPythonBase {
* For streaming execution we extract the PythonFunctionInfo from AggregateInfo.
*/
protected def extractPythonAggregateFunctionInfosFromAggregateInfo(
- pythonAggregateInfo: AggregateInfo): PythonFunctionInfo = {
+ aggIndex: Int,
+ pythonAggregateInfo: AggregateInfo): (PythonFunctionInfo, Array[DataViewSpec]) = {
pythonAggregateInfo.function match {
case function: PythonFunction =>
- new PythonFunctionInfo(
- function,
- pythonAggregateInfo.argIndexes.map(_.asInstanceOf[AnyRef]))
+ (
+ new PythonFunctionInfo(
+ function,
+ pythonAggregateInfo.argIndexes.map(_.asInstanceOf[AnyRef])),
+ extractDataViewSpecs(
+ aggIndex,
+ function.asInstanceOf[PythonAggregateFunction].getAccumulatorType)
+ )
case _: Count1AggFunction =>
// The count star will be treated specially in Python UDF worker
- PythonFunctionInfo.DUMMY_PLACEHOLDER
+ (PythonFunctionInfo.DUMMY_PLACEHOLDER, Array())
case _ =>
throw new TableException(
"Unsupported python aggregate function: " + pythonAggregateInfo.function)
}
}
+
+ protected def extractDataViewSpecs(
+ index: Int,
+ accType: TypeInformation[_]): Array[DataViewSpec] = {
+ if (!accType.isInstanceOf[CompositeType[_]]) {
+ return Array()
+ }
+
+ def includesDataView(ct: CompositeType[_]): Boolean = {
+ (0 until ct.getArity).exists(i =>
+ ct.getTypeAt(i) match {
+ case nestedCT: CompositeType[_] => includesDataView(nestedCT)
+ case t: TypeInformation[_] if t.getTypeClass == classOf[ListView[_]] => true
+ case _ => false
+ }
+ )
+ }
+
+ if (includesDataView(accType.asInstanceOf[CompositeType[_]])) {
+ accType match {
+ case rowType: RowTypeInfo =>
+ (0 until rowType.getArity).flatMap(i => {
+ rowType.getFieldTypes()(i) match {
+ case ct: CompositeType[_] if includesDataView(ct) =>
+ throw new TableException(
+ "For Python AggregateFunction DataView only supported at first " +
+ "level of accumulators of Row type.")
+ case listView: ListViewTypeInfo[_] =>
Review comment:
Side comment: This class is deprecated. Please don't use it in new code anymore. Otherwise we will never be able to drop it any time soon.
Same for `TypeConversions.fromLegacyInfoToDataType` ideally we should get rid of all these calls as quickly as possible. I'm trying to remove as many of these calls as possible but if new calls are added in PR this is an endless effort.
----------------------------------------------------------------
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 #13507: [FLINK-19231][python] Support ListState and ListView for Python UDAF.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13507:
URL: https://github.com/apache/flink/pull/13507#issuecomment-700666447
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4a0ccb37d548b600efc2d5eb5d2e786f11e6bfa2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7078",
"triggerID" : "4a0ccb37d548b600efc2d5eb5d2e786f11e6bfa2",
"triggerType" : "PUSH"
}, {
"hash" : "de6024d84022ada3b5bbdbf4c6e82a47c8459d98",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7412",
"triggerID" : "de6024d84022ada3b5bbdbf4c6e82a47c8459d98",
"triggerType" : "PUSH"
}, {
"hash" : "32df357f8334410a819301bb1a1425681185a48e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7481",
"triggerID" : "32df357f8334410a819301bb1a1425681185a48e",
"triggerType" : "PUSH"
}, {
"hash" : "958a19ec59344cf108d39153cfe25f849fed15af",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7494",
"triggerID" : "958a19ec59344cf108d39153cfe25f849fed15af",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 958a19ec59344cf108d39153cfe25f849fed15af Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7494)
<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] WeiZhong94 commented on a change in pull request #13507: [FLINK-19231][python] Support ListState and ListView for Python UDAF.
Posted by GitBox <gi...@apache.org>.
WeiZhong94 commented on a change in pull request #13507:
URL: https://github.com/apache/flink/pull/13507#discussion_r503096233
##########
File path: flink-python/pyflink/common/types.py
##########
@@ -37,7 +37,7 @@ def _create_row(fields, values, row_kind: RowKind = None):
return row
-class Row(tuple):
+class Row(object):
Review comment:
We need an updatable Row for implementing DataViews, but the tuple is read-only.
##########
File path: flink-python/pyflink/common/types.py
##########
@@ -82,21 +82,18 @@ class Row(tuple):
Row(name='Alice', age=11)
"""
- def __new__(cls, *args, **kwargs):
Review comment:
The example in the doc still works except the usage of the "in" keyword. These changes make the Row object updatable and would not break the legacy code so I think they are accpetable.
##########
File path: flink-python/pyflink/fn_execution/beam/beam_operations_slow.py
##########
@@ -304,9 +306,6 @@ def finish(self):
def reset(self):
super().reset()
- if self.keyed_state_backend:
Review comment:
The reset method would be called between the "open" method and "process_element" method of the GroupAggFunction so we should not reset the keyed_state_backend here.
----------------------------------------------------------------
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] dianfu commented on a change in pull request #13507: [FLINK-19231][python] Support ListState and ListView for Python UDAF.
Posted by GitBox <gi...@apache.org>.
dianfu commented on a change in pull request #13507:
URL: https://github.com/apache/flink/pull/13507#discussion_r497198985
##########
File path: flink-python/pyflink/common/types.py
##########
@@ -37,7 +37,7 @@ def _create_row(fields, values, row_kind: RowKind = None):
return row
-class Row(tuple):
+class Row(object):
Review comment:
Why changed this?
##########
File path: flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/python/PythonFunctionInfo.java
##########
@@ -28,7 +28,7 @@
* the actual Python function, the input arguments, etc.
*/
@Internal
-public final class PythonFunctionInfo implements Serializable {
Review comment:
Why changed this?
##########
File path: flink-python/pyflink/fn_execution/beam/beam_operations_slow.py
##########
@@ -304,9 +306,6 @@ def finish(self):
def reset(self):
super().reset()
- if self.keyed_state_backend:
Review comment:
why removing this?
##########
File path: flink-python/pyflink/proto/flink-fn-execution.proto
##########
@@ -94,6 +94,21 @@ message UserDefinedDataStreamFunctions {
// A list of the user-defined aggregate functions to be executed in a group aggregate operation.
message UserDefinedAggregateFunctions {
+ message DataViewSpec {
+ enum DataViewType {
+ LIST = 0;
+ MAP = 1;
+ }
+ string name = 1;
+ int32 field_index = 2;
+ DataViewType type = 3;
+ Schema.FieldType element_type = 4;
+ Schema.FieldType key_type = 5;
+ }
+ message DataViewSpecs {
Review comment:
Each DataViewSpec belongs to a specific UserDefinedAggregateFunction.
What about adding UserDefinedAggregateFunction which extends UserDefinedFunction and move DataViewSpec inside UserDefinedAggregateFunction?
##########
File path: flink-python/pyflink/fn_execution/aggregate.py
##########
@@ -167,18 +247,31 @@ class SimpleAggsHandleFunction(AggsHandleFunction):
def __init__(self,
udfs: List[AggregateFunction],
args_offsets_list: List[List[int]],
- index_of_count_star: int):
+ index_of_count_star: int,
+ udf_data_view_specs: List[List[DataViewSpec]]):
self._udfs = udfs
self._args_offsets_list = args_offsets_list
self._accumulators = None # type: Row
self._get_value_indexes = [i for i in range(len(udfs))]
if index_of_count_star >= 0:
# The record count is used internally, should be ignored by the get_value method.
self._get_value_indexes.remove(index_of_count_star)
+ self._udf_data_view_specs = udf_data_view_specs
+ self._udf_data_views = [] # type: List[Dict[DataView]]
Review comment:
```suggestion
self._udf_data_views = []
```
##########
File path: flink-python/pyflink/proto/flink-fn-execution.proto
##########
@@ -108,6 +123,8 @@ message UserDefinedAggregateFunctions {
int32 state_cache_size = 7;
// Cleanup the expired state if true.
bool state_cleaning_enabled = 8;
+ // The data view specifications
+ repeated DataViewSpecs udf_data_view_specs = 9;
Review comment:
```suggestion
repeated DataViewSpecs data_view_specs = 9;
```
##########
File path: flink-python/pyflink/common/types.py
##########
@@ -82,21 +82,18 @@ class Row(tuple):
Row(name='Alice', age=11)
"""
- def __new__(cls, *args, **kwargs):
Review comment:
I guess the example in the doc of the class doesn't work any more with this change. Could you explain why changed this and investigate if it's possible avoiding this change?
##########
File path: flink-python/pyflink/fn_execution/beam/beam_operations_slow.py
##########
@@ -323,6 +322,7 @@ def __init__(self, name, spec, counter_factory, sampler, consumers, keyed_state_
self.index_of_count_star = spec.serialized_fn.index_of_count_star
self.state_cache_size = spec.serialized_fn.state_cache_size
self.state_cleaning_enabled = spec.serialized_fn.state_cleaning_enabled
+ self.udf_data_view_specs = extract_data_view_specs(spec.serialized_fn.udf_data_view_specs)
Review comment:
```suggestion
self.data_view_specs = extract_data_view_specs(spec.serialized_fn.udf_data_view_specs)
```
##########
File path: flink-python/pyflink/table/data_view.py
##########
@@ -0,0 +1,100 @@
+################################################################################
+# 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.
+################################################################################
+from abc import ABC, abstractmethod
+from typing import TypeVar, Generic, Iterable, List, Any, Iterator
+
+T = TypeVar('T')
+
+__all__ = ['DataView', 'ListView']
Review comment:
What about also adding them in pyflink.table.__init__.py?
##########
File path: flink-python/pyflink/proto/flink-fn-execution.proto
##########
@@ -94,6 +94,21 @@ message UserDefinedDataStreamFunctions {
// A list of the user-defined aggregate functions to be executed in a group aggregate operation.
message UserDefinedAggregateFunctions {
+ message DataViewSpec {
+ enum DataViewType {
+ LIST = 0;
+ MAP = 1;
+ }
+ string name = 1;
+ int32 field_index = 2;
+ DataViewType type = 3;
+ Schema.FieldType element_type = 4;
Review comment:
Please split the fields for ListView and MapView, e.g.
```
message ListView {
Schema.FieldType element_type = 1;
}
message MapView {
Schema.FieldType key_type = 1;
Schema.FieldType value_type = 2;
}
message DataViewSpec {
string name = 1;
int32 field_index = 2;
oneof data_view {
ListView list_view = 3;
MapView map_view = 4;
}
}
```
##########
File path: flink-python/pyflink/fn_execution/aggregate.py
##########
@@ -258,6 +356,7 @@ def __init__(self,
aggs_handle: AggsHandleFunction,
key_selector: RowKeySelector,
state_backend: RemoteKeyedStateBackend,
+ state_value_coder: FastCoder,
Review comment:
```suggestion
state_value_coder: Coder,
```
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/typeutils/DataViewUtils.java
##########
@@ -159,7 +160,7 @@ public DataType transform(DataType dataType) {
/**
* Information about a {@link DataView} stored in state.
*/
- public abstract static class DataViewSpec {
+ public abstract static class DataViewSpec implements Serializable {
Review comment:
Add serialVersionUID for this class and the child classes?
##########
File path: flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/stream/table/PythonAggregateTest.scala
##########
@@ -60,4 +66,42 @@ class PythonAggregateTest extends TableTestBase {
util.verifyPlan(resultTable)
}
+
+ @Test
+ def testExtractDataViewSpecs(): Unit = {
+ val accTypeInfo =
+ new RowTypeInfo(Types.STRING(), new ListViewTypeInfo(Types.STRING()))
+
+ val specs = TestCommonPythonAggregate.extractDataViewSpecs(accTypeInfo)
+
+ val expected = Array(
+ new ListViewSpec("agg0$f1", 1, DataTypes.ARRAY(DataTypes.STRING())))
+
+ assertEquals(expected(0).getClass, specs(0).getClass)
+ assertEquals(expected(0).getDataType, specs(0).getDataType)
+ assertEquals(expected(0).getStateId, specs(0).getStateId)
+ assertEquals(expected(0).getFieldIndex, specs(0).getFieldIndex)
+ }
+
+ @Test(expected = classOf[TableException])
+ def testExtractSecondLevelDataViewSpecs(): Unit = {
+ val accTypeInfo =
+ new RowTypeInfo(Types.STRING(), new RowTypeInfo(new ListViewTypeInfo(Types.STRING())))
+
+ val specs = TestCommonPythonAggregate.extractDataViewSpecs(accTypeInfo)
Review comment:
```suggestion
TestCommonPythonAggregate.extractDataViewSpecs(accTypeInfo)
```
##########
File path: flink-python/pyflink/fn_execution/aggregate.py
##########
@@ -267,12 +366,12 @@ def __init__(self,
self.key_selector = key_selector
# Currently we do not support user-defined type accumulator.
Review comment:
The comments doesn't apply any more and could be removed?
##########
File path: flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/stream/table/PythonAggregateTest.scala
##########
@@ -60,4 +66,42 @@ class PythonAggregateTest extends TableTestBase {
util.verifyPlan(resultTable)
}
+
+ @Test
+ def testExtractDataViewSpecs(): Unit = {
+ val accTypeInfo =
+ new RowTypeInfo(Types.STRING(), new ListViewTypeInfo(Types.STRING()))
+
+ val specs = TestCommonPythonAggregate.extractDataViewSpecs(accTypeInfo)
+
+ val expected = Array(
+ new ListViewSpec("agg0$f1", 1, DataTypes.ARRAY(DataTypes.STRING())))
+
+ assertEquals(expected(0).getClass, specs(0).getClass)
+ assertEquals(expected(0).getDataType, specs(0).getDataType)
+ assertEquals(expected(0).getStateId, specs(0).getStateId)
+ assertEquals(expected(0).getFieldIndex, specs(0).getFieldIndex)
+ }
+
+ @Test(expected = classOf[TableException])
+ def testExtractSecondLevelDataViewSpecs(): Unit = {
+ val accTypeInfo =
+ new RowTypeInfo(Types.STRING(), new RowTypeInfo(new ListViewTypeInfo(Types.STRING())))
+
+ val specs = TestCommonPythonAggregate.extractDataViewSpecs(accTypeInfo)
+ }
+
+ @Test(expected = classOf[TableException])
+ def testExtractDataViewSpecsFromTupleType(): Unit = {
+ val accTypeInfo =
+ new TupleTypeInfo(Types.STRING(), new ListViewTypeInfo(Types.STRING()))
+
+ val specs = TestCommonPythonAggregate.extractDataViewSpecs(accTypeInfo)
Review comment:
```suggestion
TestCommonPythonAggregate.extractDataViewSpecs(accTypeInfo)
```
----------------------------------------------------------------
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 #13507: [FLINK-19231][python] Support ListState and ListView for Python UDAF.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13507:
URL: https://github.com/apache/flink/pull/13507#issuecomment-700666447
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4a0ccb37d548b600efc2d5eb5d2e786f11e6bfa2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7078",
"triggerID" : "4a0ccb37d548b600efc2d5eb5d2e786f11e6bfa2",
"triggerType" : "PUSH"
}, {
"hash" : "de6024d84022ada3b5bbdbf4c6e82a47c8459d98",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7412",
"triggerID" : "de6024d84022ada3b5bbdbf4c6e82a47c8459d98",
"triggerType" : "PUSH"
}, {
"hash" : "32df357f8334410a819301bb1a1425681185a48e",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7481",
"triggerID" : "32df357f8334410a819301bb1a1425681185a48e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* de6024d84022ada3b5bbdbf4c6e82a47c8459d98 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7412)
* 32df357f8334410a819301bb1a1425681185a48e Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7481)
<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 #13507: [FLINK-19231][python] Support ListState and ListView for Python UDAF.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13507:
URL: https://github.com/apache/flink/pull/13507#issuecomment-700666447
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4a0ccb37d548b600efc2d5eb5d2e786f11e6bfa2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7078",
"triggerID" : "4a0ccb37d548b600efc2d5eb5d2e786f11e6bfa2",
"triggerType" : "PUSH"
}, {
"hash" : "de6024d84022ada3b5bbdbf4c6e82a47c8459d98",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7412",
"triggerID" : "de6024d84022ada3b5bbdbf4c6e82a47c8459d98",
"triggerType" : "PUSH"
}, {
"hash" : "32df357f8334410a819301bb1a1425681185a48e",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7481",
"triggerID" : "32df357f8334410a819301bb1a1425681185a48e",
"triggerType" : "PUSH"
}, {
"hash" : "958a19ec59344cf108d39153cfe25f849fed15af",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "958a19ec59344cf108d39153cfe25f849fed15af",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* de6024d84022ada3b5bbdbf4c6e82a47c8459d98 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7412)
* 32df357f8334410a819301bb1a1425681185a48e Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7481)
* 958a19ec59344cf108d39153cfe25f849fed15af 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] WeiZhong94 commented on pull request #13507: [FLINK-19231][python] Support ListState and ListView for Python UDAF.
Posted by GitBox <gi...@apache.org>.
WeiZhong94 commented on pull request #13507:
URL: https://github.com/apache/flink/pull/13507#issuecomment-706940408
@dianfu @twalthr Thanks for your comments. I have updated 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