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/10/22 04:59:03 UTC
[GitHub] [flink] WeiZhong94 opened a new pull request #13739: [FLINK-19232][python] support iterating MapState and MapView
WeiZhong94 opened a new pull request #13739:
URL: https://github.com/apache/flink/pull/13739
## What is the purpose of the change
*This pull request supports iterating MapState and MapView*
## Brief change log
- *Support iterating MapState and MapView*
## 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)`: (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? (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] flinkbot edited a comment on pull request #13739: [FLINK-19232][python] support iterating MapState and MapView
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13739:
URL: https://github.com/apache/flink/pull/13739#issuecomment-714232634
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9f75014b48308eb93937e11fcd8f4da80de12bd4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8074",
"triggerID" : "9f75014b48308eb93937e11fcd8f4da80de12bd4",
"triggerType" : "PUSH"
}, {
"hash" : "844e71b94546c009d1e0e5ec7fccf841354f151e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8112",
"triggerID" : "844e71b94546c009d1e0e5ec7fccf841354f151e",
"triggerType" : "PUSH"
}, {
"hash" : "bc25118caf893bd8b367649fee8d7c9c8ee8dfd4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8140",
"triggerID" : "bc25118caf893bd8b367649fee8d7c9c8ee8dfd4",
"triggerType" : "PUSH"
}, {
"hash" : "2b7a1fc6e433f2d2ccc0944d11db9b3da2705cf2",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8311",
"triggerID" : "2b7a1fc6e433f2d2ccc0944d11db9b3da2705cf2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2b7a1fc6e433f2d2ccc0944d11db9b3da2705cf2 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8311)
<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 #13739: [FLINK-19232][python] support iterating MapState and MapView
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13739:
URL: https://github.com/apache/flink/pull/13739#issuecomment-714232634
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9f75014b48308eb93937e11fcd8f4da80de12bd4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8074",
"triggerID" : "9f75014b48308eb93937e11fcd8f4da80de12bd4",
"triggerType" : "PUSH"
}, {
"hash" : "844e71b94546c009d1e0e5ec7fccf841354f151e",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8112",
"triggerID" : "844e71b94546c009d1e0e5ec7fccf841354f151e",
"triggerType" : "PUSH"
}, {
"hash" : "bc25118caf893bd8b367649fee8d7c9c8ee8dfd4",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8140",
"triggerID" : "bc25118caf893bd8b367649fee8d7c9c8ee8dfd4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 844e71b94546c009d1e0e5ec7fccf841354f151e Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8112)
* bc25118caf893bd8b367649fee8d7c9c8ee8dfd4 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8140)
<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 #13739: [FLINK-19232][python] support iterating MapState and MapView
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13739:
URL: https://github.com/apache/flink/pull/13739#issuecomment-714232634
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9f75014b48308eb93937e11fcd8f4da80de12bd4",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8074",
"triggerID" : "9f75014b48308eb93937e11fcd8f4da80de12bd4",
"triggerType" : "PUSH"
}, {
"hash" : "844e71b94546c009d1e0e5ec7fccf841354f151e",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "844e71b94546c009d1e0e5ec7fccf841354f151e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 9f75014b48308eb93937e11fcd8f4da80de12bd4 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8074)
* 844e71b94546c009d1e0e5ec7fccf841354f151e 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] dianfu closed pull request #13739: [FLINK-19232][python] support iterating MapState and MapView
Posted by GitBox <gi...@apache.org>.
dianfu closed pull request #13739:
URL: https://github.com/apache/flink/pull/13739
----------------------------------------------------------------
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 #13739: [FLINK-19232][python] support iterating MapState and MapView
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13739:
URL: https://github.com/apache/flink/pull/13739#issuecomment-714232634
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9f75014b48308eb93937e11fcd8f4da80de12bd4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8074",
"triggerID" : "9f75014b48308eb93937e11fcd8f4da80de12bd4",
"triggerType" : "PUSH"
}, {
"hash" : "844e71b94546c009d1e0e5ec7fccf841354f151e",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8112",
"triggerID" : "844e71b94546c009d1e0e5ec7fccf841354f151e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 844e71b94546c009d1e0e5ec7fccf841354f151e Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8112)
<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 #13739: [FLINK-19232][python] support iterating MapState and MapView
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13739:
URL: https://github.com/apache/flink/pull/13739#issuecomment-714232634
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9f75014b48308eb93937e11fcd8f4da80de12bd4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8074",
"triggerID" : "9f75014b48308eb93937e11fcd8f4da80de12bd4",
"triggerType" : "PUSH"
}, {
"hash" : "844e71b94546c009d1e0e5ec7fccf841354f151e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8112",
"triggerID" : "844e71b94546c009d1e0e5ec7fccf841354f151e",
"triggerType" : "PUSH"
}, {
"hash" : "bc25118caf893bd8b367649fee8d7c9c8ee8dfd4",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8140",
"triggerID" : "bc25118caf893bd8b367649fee8d7c9c8ee8dfd4",
"triggerType" : "PUSH"
}, {
"hash" : "2b7a1fc6e433f2d2ccc0944d11db9b3da2705cf2",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8311",
"triggerID" : "2b7a1fc6e433f2d2ccc0944d11db9b3da2705cf2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* bc25118caf893bd8b367649fee8d7c9c8ee8dfd4 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8140)
* 2b7a1fc6e433f2d2ccc0944d11db9b3da2705cf2 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8311)
<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 #13739: [FLINK-19232][python] support iterating MapState and MapView
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #13739:
URL: https://github.com/apache/flink/pull/13739#issuecomment-714232634
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9f75014b48308eb93937e11fcd8f4da80de12bd4",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "9f75014b48308eb93937e11fcd8f4da80de12bd4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 9f75014b48308eb93937e11fcd8f4da80de12bd4 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 #13739: [FLINK-19232][python] support iterating MapState and MapView
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13739:
URL: https://github.com/apache/flink/pull/13739#issuecomment-714232634
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9f75014b48308eb93937e11fcd8f4da80de12bd4",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8074",
"triggerID" : "9f75014b48308eb93937e11fcd8f4da80de12bd4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 9f75014b48308eb93937e11fcd8f4da80de12bd4 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8074)
<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 #13739: [FLINK-19232][python] support iterating MapState and MapView
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13739:
URL: https://github.com/apache/flink/pull/13739#issuecomment-714232634
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9f75014b48308eb93937e11fcd8f4da80de12bd4",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8074",
"triggerID" : "9f75014b48308eb93937e11fcd8f4da80de12bd4",
"triggerType" : "PUSH"
}, {
"hash" : "844e71b94546c009d1e0e5ec7fccf841354f151e",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8112",
"triggerID" : "844e71b94546c009d1e0e5ec7fccf841354f151e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 9f75014b48308eb93937e11fcd8f4da80de12bd4 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8074)
* 844e71b94546c009d1e0e5ec7fccf841354f151e Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8112)
<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 #13739: [FLINK-19232][python] support iterating MapState and MapView
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13739:
URL: https://github.com/apache/flink/pull/13739#issuecomment-714232634
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9f75014b48308eb93937e11fcd8f4da80de12bd4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8074",
"triggerID" : "9f75014b48308eb93937e11fcd8f4da80de12bd4",
"triggerType" : "PUSH"
}, {
"hash" : "844e71b94546c009d1e0e5ec7fccf841354f151e",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8112",
"triggerID" : "844e71b94546c009d1e0e5ec7fccf841354f151e",
"triggerType" : "PUSH"
}, {
"hash" : "bc25118caf893bd8b367649fee8d7c9c8ee8dfd4",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "bc25118caf893bd8b367649fee8d7c9c8ee8dfd4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 844e71b94546c009d1e0e5ec7fccf841354f151e Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8112)
* bc25118caf893bd8b367649fee8d7c9c8ee8dfd4 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 #13739: [FLINK-19232][python] support iterating MapState and MapView
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13739:
URL: https://github.com/apache/flink/pull/13739#issuecomment-714232634
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9f75014b48308eb93937e11fcd8f4da80de12bd4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8074",
"triggerID" : "9f75014b48308eb93937e11fcd8f4da80de12bd4",
"triggerType" : "PUSH"
}, {
"hash" : "844e71b94546c009d1e0e5ec7fccf841354f151e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8112",
"triggerID" : "844e71b94546c009d1e0e5ec7fccf841354f151e",
"triggerType" : "PUSH"
}, {
"hash" : "bc25118caf893bd8b367649fee8d7c9c8ee8dfd4",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8140",
"triggerID" : "bc25118caf893bd8b367649fee8d7c9c8ee8dfd4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* bc25118caf893bd8b367649fee8d7c9c8ee8dfd4 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8140)
<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 #13739: [FLINK-19232][python] support iterating MapState and MapView
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13739:
URL: https://github.com/apache/flink/pull/13739#issuecomment-714232634
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9f75014b48308eb93937e11fcd8f4da80de12bd4",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8074",
"triggerID" : "9f75014b48308eb93937e11fcd8f4da80de12bd4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 9f75014b48308eb93937e11fcd8f4da80de12bd4 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8074)
<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 #13739: [FLINK-19232][python] support iterating MapState and MapView
Posted by GitBox <gi...@apache.org>.
dianfu commented on a change in pull request #13739:
URL: https://github.com/apache/flink/pull/13739#discussion_r509978593
##########
File path: flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/BeamPythonFunctionRunner.java
##########
@@ -863,6 +965,84 @@ private static StateRequestHandler getStateRequestHandler(
}
}
+ private BeamFnApi.StateGetResponse.Builder handleMapIterateRequest(
+ MapState<ByteArrayWrapper, byte[]> mapState,
+ IterateType iterateType,
+ ByteArrayWrapper iteratorToken) throws Exception {
+ final Iterator iterator;
+ if (iteratorToken == null) {
+ switch (iterateType) {
+ case ITEMS:
+ case VALUES:
+ iterator = mapState.iterator();
+ break;
+ case KEYS:
+ iterator = mapState.keys().iterator();
+ break;
+ default:
+ throw new RuntimeException("Unsupported iterate type: " + iterateType);
+ }
+ } else {
+ iterator = mapStateIteratorCache.get(iteratorToken);
+ if (iterator == null) {
+ throw new RuntimeException("The cached iterator not exist!");
Review comment:
```suggestion
throw new RuntimeException("The cached iterator does not exist!");
```
##########
File path: flink-python/pyflink/fn_execution/state_impl.py
##########
@@ -301,6 +380,124 @@ def _convert_to_cache_key(state_key):
return state_key.SerializeToString()
+class RemovableIterator(collections.Iterator):
+
+ def __init__(self, internal_map_state, iterate_type):
+ self._internal_map_state = internal_map_state
+ self._mod_count = internal_map_state._mod_count
+ self._underlying = iter(self._internal_map_state._write_cache.items())
+ self._underlying_is_write_cache = True
+ self._underlying_is_read_cache = False
+ self._iterator_token = None
+ self._cached_map_state = None
+ self._last_key = None
+ self._iterate_type = iterate_type
+ self._removed_keys = set()
+ if self._iterate_type == IterateType.KEYS:
+ self._get_from_cache = self._get_key_from_cache
+ self._get_from_data = self._get_key_from_data
+ elif self._iterate_type == IterateType.VALUES:
+ self._get_from_cache = self._get_value_from_cache
+ self._get_from_data = self._get_value_from_data
+ else:
+ self._get_from_cache = self._get_item_from_cache
+ self._get_from_data = self._get_item_from_data
+
+ def __next__(self):
+ self._check_modification()
+ if self._underlying_is_write_cache:
+ # Iterate the data in write cache firstly
+ try:
+ key, existed_and_value = next(self._underlying)
+ while not existed_and_value[0]:
+ key, existed_and_value = next(self._underlying)
+ return self._get_from_cache(key, existed_and_value)
+ except StopIteration:
+ self._underlying = self._next_batch()
+ self._underlying_is_write_cache = False
+ if self._underlying_is_read_cache:
+ # If the read cache contains all data (except the data in write cache) of the map state
+ self._last_key, existed_and_value = next(self._underlying)
+ while not existed_and_value[0] or \
+ self._last_key in self._internal_map_state._write_cache:
+ key, existed_and_value = next(self._underlying)
+ return self._get_from_cache(self._last_key, existed_and_value)
+ else:
+ try:
+ # The data is from Java side
+ next_value = self._get_from_data(next(self._underlying))
+ while self._last_key in self._internal_map_state._write_cache:
+ next_value = self._get_from_data(next(self._underlying))
+ return next_value
+ except StopIteration:
+ if self._iterator_token is None:
+ raise
+ else:
+ self._underlying = self._next_batch()
+ return self.__next__()
+
+ def remove(self):
+ """
+ Remove the the last element returned by this iterator.
+ """
+ if self._last_key is None:
+ raise Exception("You need to call the '__next__' method before calling "
+ "this method.")
+ self._check_modification()
+ # Bypass the 'remove' method of the map state to avoid triggering the commit of the write
+ # cache.
+ if self._internal_map_state._cleared:
+ del self._internal_map_state._write_cache[self._last_key]
+ self._mod_count += 1
Review comment:
Could refactor a bit to remove the duplicate code
##########
File path: flink-python/pyflink/fn_execution/state_impl.py
##########
@@ -301,6 +380,124 @@ def _convert_to_cache_key(state_key):
return state_key.SerializeToString()
+class RemovableIterator(collections.Iterator):
+
+ def __init__(self, internal_map_state, iterate_type):
+ self._internal_map_state = internal_map_state
+ self._mod_count = internal_map_state._mod_count
+ self._underlying = iter(self._internal_map_state._write_cache.items())
+ self._underlying_is_write_cache = True
+ self._underlying_is_read_cache = False
+ self._iterator_token = None
+ self._cached_map_state = None
+ self._last_key = None
+ self._iterate_type = iterate_type
+ self._removed_keys = set()
+ if self._iterate_type == IterateType.KEYS:
+ self._get_from_cache = self._get_key_from_cache
+ self._get_from_data = self._get_key_from_data
+ elif self._iterate_type == IterateType.VALUES:
+ self._get_from_cache = self._get_value_from_cache
+ self._get_from_data = self._get_value_from_data
+ else:
+ self._get_from_cache = self._get_item_from_cache
+ self._get_from_data = self._get_item_from_data
+
+ def __next__(self):
+ self._check_modification()
+ if self._underlying_is_write_cache:
+ # Iterate the data in write cache firstly
+ try:
+ key, existed_and_value = next(self._underlying)
Review comment:
```suggestion
self._last_key, (exists, value) = next(self._underlying)
```
##########
File path: flink-python/src/main/java/org/apache/flink/python/PythonOptions.java
##########
@@ -164,11 +164,22 @@
*/
@Experimental
public static final ConfigOption<Integer> MAP_STATE_WRITE_CACHE_SIZE = ConfigOptions
- .key("python.map-state.write.cache.size")
+ .key("python.map-state.write-cache-size")
.defaultValue(1000)
.withDescription("The maximum number of cached write requests for a single Python " +
"MapState. The write requests will be flushed to the state backend (managed in " +
"the Java operator) when the number of cached write requests exceed this limit. " +
"Note that this is an experimental flag and might not be available in future " +
"releases.");
+
+ /**
+ * The maximum number of write requests cached in a Python MapState.
+ */
+ @Experimental
+ public static final ConfigOption<Integer> MAP_STATE_ITERATE_CACHE_SIZE = ConfigOptions
+ .key("python.map-state.iterate-cache-size")
+ .defaultValue(1000)
+ .withDescription("The maximum number of entries read from Java side when iterating a " +
Review comment:
What about improving the doc a bit to make it more clear what this config option is used for?
##########
File path: flink-python/pyflink/fn_execution/state_impl.py
##########
@@ -395,6 +619,17 @@ def commit(self):
self._write_cache.clear()
self._cleared = False
+ def _iterate_next_batch(self, iterate_type, iterator_token):
Review comment:
```suggestion
def next_batch(self, iterate_type, iterator_token):
```
##########
File path: docs/_includes/generated/python_configuration.html
##########
@@ -57,13 +57,19 @@
<td>If set, the Python worker will configure itself to use the managed memory budget of the task slot. Otherwise, it will use the Off-Heap Memory of the task slot. In this case, users should set the Task Off-Heap Memory using the configuration key taskmanager.memory.task.off-heap.size.</td>
</tr>
<tr>
- <td><h5>python.map-state.read.cache.size</h5></td>
+ <td><h5>python.map-state.iterate-cache-size</h5></td>
Review comment:
python.map-state.iterate-req-batch-size
----------------------------------------------------------------
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 #13739: [FLINK-19232][python] support iterating MapState and MapView
Posted by GitBox <gi...@apache.org>.
WeiZhong94 commented on pull request #13739:
URL: https://github.com/apache/flink/pull/13739#issuecomment-714486868
@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 #13739: [FLINK-19232][python] support iterating MapState and MapView
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13739:
URL: https://github.com/apache/flink/pull/13739#issuecomment-714232634
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9f75014b48308eb93937e11fcd8f4da80de12bd4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8074",
"triggerID" : "9f75014b48308eb93937e11fcd8f4da80de12bd4",
"triggerType" : "PUSH"
}, {
"hash" : "844e71b94546c009d1e0e5ec7fccf841354f151e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8112",
"triggerID" : "844e71b94546c009d1e0e5ec7fccf841354f151e",
"triggerType" : "PUSH"
}, {
"hash" : "bc25118caf893bd8b367649fee8d7c9c8ee8dfd4",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8140",
"triggerID" : "bc25118caf893bd8b367649fee8d7c9c8ee8dfd4",
"triggerType" : "PUSH"
}, {
"hash" : "2b7a1fc6e433f2d2ccc0944d11db9b3da2705cf2",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "2b7a1fc6e433f2d2ccc0944d11db9b3da2705cf2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* bc25118caf893bd8b367649fee8d7c9c8ee8dfd4 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8140)
* 2b7a1fc6e433f2d2ccc0944d11db9b3da2705cf2 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #13739: [FLINK-19232][python] support iterating MapState and MapView
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #13739:
URL: https://github.com/apache/flink/pull/13739#issuecomment-714227770
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 9f75014b48308eb93937e11fcd8f4da80de12bd4 (Thu Oct 22 05:01:49 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