You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "dmvk (via GitHub)" <gi...@apache.org> on 2024/01/12 09:49:16 UTC
[PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
dmvk opened a new pull request, #24079:
URL: https://github.com/apache/flink/pull/24079
https://issues.apache.org/jira/browse/FLINK-34063
--
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
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "echauchot (via GitHub)" <gi...@apache.org>.
echauchot commented on PR #24079:
URL: https://github.com/apache/flink/pull/24079#issuecomment-1911828089
I finally had time to take a look at this PR (reduced availability nowadays). Thanks for all the diagnostics and chained fixes @dmvk and sorry for missing the fact that pos was not updated unless underlying snappy buffer is flushed. I should have added more test coverage for sure ! Thanks to the reviewers as well !
--
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
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "dmvk (via GitHub)" <gi...@apache.org>.
dmvk commented on PR #24079:
URL: https://github.com/apache/flink/pull/24079#issuecomment-1895414707
> it also solves the the compressed representation might actually end up being bigger than uncompressed one problem.
Not completely, since we're doing per-value compression for list states, (I'm not 100% sure how snappy works, assuming something along the lines of huffman coding) you still need to write per-frame metadata, so it could imo still be bigger for really small values. It certainly makes things better though.
--
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
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on PR #24079:
URL: https://github.com/apache/flink/pull/24079#issuecomment-1893527794
> I've been able to pin down the issue to https://github.com/apache/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorStateRestoreOperation.java#L190
>
> the problem is that the sorting breaks on empty state ... we should always restore in the order state was written
>
> fix in: [8d02807](https://github.com/apache/flink/commit/8d02807e3b8b12c010cde64e2518fe49044eda14)
>
> cc @fredia @ruibinx since you've worked on #23938
@dmvk Thanks for the investigation and fix, sorry for ignoring the case of `length=0` https://github.com/apache/flink/pull/23938#discussion_r1432321260 at that time.
> It seems that we should not really have snappy headers per-state, but we should write them just once at the begining ...
I have an inelegant solution: before deserializing each state, record the `startPos` first, and then when constructing `CompressibleFSDataInputStream` each time, seek to `startPos` first, so that the order of our offset is no longer important, and compatibility does not need to be specially considered.
```java
void restore() {
for (OperatorStateHandle stateHandle : stateHandles) {
...
long startPos = in.getPos();
for (String stateName : toRestore) {
in.seek(startPos);
try (final CompressibleFSDataInputStream compressedIn =
new CompressibleFSDataInputStream(
in,
compressionDecorator)) {
// deserialize
}
}
...
}
}
```
--
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
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "dmvk (via GitHub)" <gi...@apache.org>.
dmvk commented on PR #24079:
URL: https://github.com/apache/flink/pull/24079#issuecomment-1888797223
fyi @echauchot @dawidwys
--
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
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "isburmistrov (via GitHub)" <gi...@apache.org>.
isburmistrov commented on code in PR #24079:
URL: https://github.com/apache/flink/pull/24079#discussion_r1450346151
##########
flink-runtime/src/main/java/org/apache/flink/runtime/state/CompressibleFSDataOutputStream.java:
##########
@@ -41,6 +41,10 @@ public CompressibleFSDataOutputStream(
@Override
public long getPos() throws IOException {
+ // Underlying compression involves buffering, so the only way to report correct position is
+ // to flush the underlying stream. This lowers the effectivity of compression, but there is
+ // no other way, since the position is often used as a split point.
+ flush();
Review Comment:
Also, FYI, read operation in `SnappyFramedInputStream` also does the buffering:
```
@Override
public int read()
throws IOException
{
if (closed) {
return -1;
}
if (!ensureBuffer()) {
return -1;
}
return buffer[position++] & 0xFF;
}
```
So when we do `seek` in the `CompressibleFSDataInputStream` (which gets delegated to `delegate`) and then do `read` (which gets delegated to `compressingDelegate`) - I guess something weird can happen.
--
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
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "dmvk (via GitHub)" <gi...@apache.org>.
dmvk commented on code in PR #24079:
URL: https://github.com/apache/flink/pull/24079#discussion_r1450722556
##########
flink-runtime/src/main/java/org/apache/flink/runtime/state/CompressibleFSDataOutputStream.java:
##########
@@ -41,6 +41,10 @@ public CompressibleFSDataOutputStream(
@Override
public long getPos() throws IOException {
+ // Underlying compression involves buffering, so the only way to report correct position is
+ // to flush the underlying stream. This lowers the effectivity of compression, but there is
+ // no other way, since the position is often used as a split point.
+ flush();
Review Comment:
Your partially right. One missing point is that snappy always buffers at most until the next flush, so as long as we always read full record, it should work as expected. In case of partial reads, it works as you describe.
I've tried to fix the partial read case in https://github.com/apache/flink/pull/24079/commits/4a42f16a13f3d22c188a5020a8e8718394d11e0c, PTAL
--
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
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "echauchot (via GitHub)" <gi...@apache.org>.
echauchot commented on PR #24079:
URL: https://github.com/apache/flink/pull/24079#issuecomment-1893752215
> fyi @echauchot @dawidwys
Sorry for introducing this bug and thanks for fixing @dmvk. I'll look at it soon.
--
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
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "dmvk (via GitHub)" <gi...@apache.org>.
dmvk commented on PR #24079:
URL: https://github.com/apache/flink/pull/24079#issuecomment-1893253680
> Yeah exactly. I wonder - is it hard to simply not do compression for partitioned state (but keep doing it for broadcast state)?
It requires changing the wire format.
--
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
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "dmvk (via GitHub)" <gi...@apache.org>.
dmvk commented on PR #24079:
URL: https://github.com/apache/flink/pull/24079#issuecomment-1892793322
And another one ... this just hurts :/
`DataSetSavepointReaderITCase.testOperatorStateInputFormat`
--
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
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "isburmistrov (via GitHub)" <gi...@apache.org>.
isburmistrov commented on code in PR #24079:
URL: https://github.com/apache/flink/pull/24079#discussion_r1450246875
##########
flink-runtime/src/main/java/org/apache/flink/runtime/state/CompressibleFSDataOutputStream.java:
##########
@@ -41,6 +41,10 @@ public CompressibleFSDataOutputStream(
@Override
public long getPos() throws IOException {
+ // Underlying compression involves buffering, so the only way to report correct position is
+ // to flush the underlying stream. This lowers the effectivity of compression, but there is
+ // no other way, since the position is often used as a split point.
+ flush();
Review Comment:
Do you think it's the best way to fix it? I think this approach has a few drawbacks:
- As mentioned, lower efficiency of compression. Actually for some compression algorithms this may make it useless, if not harmful (if there is some significant amount of metadata that gets written on `flush`)
- It's somewhat strange that the result is different whether we call `getPos()` in the process of compressing the output, or we don't. This may later come as a big surprise for someone using `CompressibleFSDataOutputStream`.
- `Seek` operation in the corresponding `CompressibleFSDataInputStream` doesn't make sense for any position, only for some specific positions. Which is also not super bad, but confusing.
IMO, a better approach may be to change the semantics of `getPos()` for `CompressibleFSDataOutputStream`. This can represent the position in the **uncompressed** data.
Correspondingly, `seek` operation in `CompressibleFSDataInputStream` can be implemented having this semantics in mind: one can read the `compressingDelegate` until the desired position is reached.
This approach also doesn't have the problem of `seek` being valid for specific positions only, which is a nice side effect.
--
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
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "dmvk (via GitHub)" <gi...@apache.org>.
dmvk commented on PR #24079:
URL: https://github.com/apache/flink/pull/24079#issuecomment-1895445564
There might actually be something more fancy going on, need to read up on it at some point :D https://en.wikipedia.org/wiki/Snappy_(compression)
--
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
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "dmvk (via GitHub)" <gi...@apache.org>.
dmvk commented on PR #24079:
URL: https://github.com/apache/flink/pull/24079#issuecomment-1894085294
> I have an inelegant solution:
@fredia After https://github.com/apache/flink/pull/24079/commits/6bb32b114b76515757e602612dcf33668d031fef the sorting is no longer important, because that allows us to seek wherever we want. It's still a nice optimization to have though, since it will allow us to read things sequentially, but ultimately we could remove it now.
Any thoughts?
--
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
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "isburmistrov (via GitHub)" <gi...@apache.org>.
isburmistrov commented on code in PR #24079:
URL: https://github.com/apache/flink/pull/24079#discussion_r1450737039
##########
flink-runtime/src/main/java/org/apache/flink/runtime/state/CompressibleFSDataOutputStream.java:
##########
@@ -41,6 +41,10 @@ public CompressibleFSDataOutputStream(
@Override
public long getPos() throws IOException {
+ // Underlying compression involves buffering, so the only way to report correct position is
+ // to flush the underlying stream. This lowers the effectivity of compression, but there is
+ // no other way, since the position is often used as a split point.
+ flush();
Review Comment:
Great, thanks for additional testing!
--
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
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #24079:
URL: https://github.com/apache/flink/pull/24079#issuecomment-1888781951
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0ea3238afa0d7da0e52f5065fe3c469585fb6b03",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "0ea3238afa0d7da0e52f5065fe3c469585fb6b03",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0ea3238afa0d7da0e52f5065fe3c469585fb6b03 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "dmvk (via GitHub)" <gi...@apache.org>.
dmvk commented on PR #24079:
URL: https://github.com/apache/flink/pull/24079#issuecomment-1892836002
hmm, so the issue is related to repartitioning in case there are multiple states;
imagine we have two states with 6 elements and we rescale from 1 to 3; this means that every split gets 2 elements;
1) we read 2 elements from first state
2) now we want to read 2 elements from 2nd state, but we're resuming read from where we left off (before seeking to offset) and surprise, headers are not there ๐
It seems that we should not really have snappy headers per-state, but we should write them just once at the begining ...
--
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
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "isburmistrov (via GitHub)" <gi...@apache.org>.
isburmistrov commented on code in PR #24079:
URL: https://github.com/apache/flink/pull/24079#discussion_r1450506731
##########
flink-runtime/src/main/java/org/apache/flink/runtime/state/CompressibleFSDataOutputStream.java:
##########
@@ -41,6 +41,10 @@ public CompressibleFSDataOutputStream(
@Override
public long getPos() throws IOException {
+ // Underlying compression involves buffering, so the only way to report correct position is
+ // to flush the underlying stream. This lowers the effectivity of compression, but there is
+ // no other way, since the position is often used as a split point.
+ flush();
Review Comment:
What are your thoughts on my second comment @dmvk? It's still an issue, isn't it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "isburmistrov (via GitHub)" <gi...@apache.org>.
isburmistrov commented on code in PR #24079:
URL: https://github.com/apache/flink/pull/24079#discussion_r1450546553
##########
flink-runtime/src/main/java/org/apache/flink/runtime/state/CompressibleFSDataOutputStream.java:
##########
@@ -41,6 +41,10 @@ public CompressibleFSDataOutputStream(
@Override
public long getPos() throws IOException {
+ // Underlying compression involves buffering, so the only way to report correct position is
+ // to flush the underlying stream. This lowers the effectivity of compression, but there is
+ // no other way, since the position is often used as a split point.
+ flush();
Review Comment:
This seems to be more hidden, but as we stick with correct semantics of Stream interface should be addressed. I believe the following scenario should reveal it:
- Create `CompressibleFSDataInputStream` with snappy
- Seek to 1st item
- Read it (will be read correctly)
- Seek to say 5th item
- Read it (I believe it will be read incorrectly)
The main point is that the 2nd seek should be to anything except to 2nd element. 2nd element will be read correctly, but anything else not - due to Snappy stream reading data from the buffer.
--
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
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "dmvk (via GitHub)" <gi...@apache.org>.
dmvk commented on PR #24079:
URL: https://github.com/apache/flink/pull/24079#issuecomment-1892871387
I'm actually starting to doubt whether the feature is even worth it for partitioned state. The compressed representation might actually end up being bigger than uncompressed one. To make it worth something, we should probably evolve partitioned state to use spare index (still, it's a separate effort).
In terms of broadcast state, this makes a perfect sense, because there we just treat everything as a single block for compression.
--
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
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on PR #24079:
URL: https://github.com/apache/flink/pull/24079#issuecomment-1894846863
> but ultimately we could remove it now.
Yes, I think [acad43f](https://github.com/apache/flink/pull/24079/commits/acad43fa1458d4544840a7b28e2e752056e5e504) is a more thorough solution, and as you mentioned, it also solves the `the compressed representation might actually end up being bigger than uncompressed one` problem.
> I don't think that needs to be considered at all, since the states are corrupted anyway, because the lack of flushes.
Agree๐, all previously generated compressed operator states are unrecoverable. Thanks to [decaf21](https://github.com/apache/flink/pull/24079/commits/decaf21efcad6c8bc4367192b2ecedd7cbd9d021), make the problems related to the compression operation state expose.
--
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
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "dmvk (via GitHub)" <gi...@apache.org>.
dmvk commented on PR #24079:
URL: https://github.com/apache/flink/pull/24079#issuecomment-1888812016
CI seems broken by https://github.com/apache/flink/pull/24022#issuecomment-1888811191
--
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
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "isburmistrov (via GitHub)" <gi...@apache.org>.
isburmistrov commented on code in PR #24079:
URL: https://github.com/apache/flink/pull/24079#discussion_r1450546553
##########
flink-runtime/src/main/java/org/apache/flink/runtime/state/CompressibleFSDataOutputStream.java:
##########
@@ -41,6 +41,10 @@ public CompressibleFSDataOutputStream(
@Override
public long getPos() throws IOException {
+ // Underlying compression involves buffering, so the only way to report correct position is
+ // to flush the underlying stream. This lowers the effectivity of compression, but there is
+ // no other way, since the position is often used as a split point.
+ flush();
Review Comment:
This issue is a bit more hidden, but as we stick with correct semantics of Stream interface should be addressed. I believe the following scenario should reveal it:
- Create `CompressibleFSDataInputStream` with snappy
- Seek to 1st item
- Read it (will be read correctly)
- Seek to say 5th item
- Read it (I believe it will be read incorrectly)
The main point is that the 2nd seek should be to anything except to 2nd element. 2nd element will be read correctly, but anything else not - due to Snappy stream reading data from the buffer.
--
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
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "isburmistrov (via GitHub)" <gi...@apache.org>.
isburmistrov commented on code in PR #24079:
URL: https://github.com/apache/flink/pull/24079#discussion_r1450546553
##########
flink-runtime/src/main/java/org/apache/flink/runtime/state/CompressibleFSDataOutputStream.java:
##########
@@ -41,6 +41,10 @@ public CompressibleFSDataOutputStream(
@Override
public long getPos() throws IOException {
+ // Underlying compression involves buffering, so the only way to report correct position is
+ // to flush the underlying stream. This lowers the effectivity of compression, but there is
+ // no other way, since the position is often used as a split point.
+ flush();
Review Comment:
This issue is a bit more hidden, but as we stick with correct semantics of Stream interface should be addressed. I believe the following scenario should reveal it:
- Create `CompressibleFSDataInputStream` with snappy
- Seek to 1st item
- Read it (will be read correctly)
- Seek to say 5th item
- Read it (I believe it will be read incorrectly)
The main point is that the 2nd seek should be outside of the underlying buffer in the snappy stream.
--
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
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "dmvk (via GitHub)" <gi...@apache.org>.
dmvk commented on PR #24079:
URL: https://github.com/apache/flink/pull/24079#issuecomment-1890348927
`testOperatorEventLostWithReaderFailure` ... reliably fails when trying to restore from compressed state, will investigate further :/
--
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
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "isburmistrov (via GitHub)" <gi...@apache.org>.
isburmistrov commented on code in PR #24079:
URL: https://github.com/apache/flink/pull/24079#discussion_r1450246875
##########
flink-runtime/src/main/java/org/apache/flink/runtime/state/CompressibleFSDataOutputStream.java:
##########
@@ -41,6 +41,10 @@ public CompressibleFSDataOutputStream(
@Override
public long getPos() throws IOException {
+ // Underlying compression involves buffering, so the only way to report correct position is
+ // to flush the underlying stream. This lowers the effectivity of compression, but there is
+ // no other way, since the position is often used as a split point.
+ flush();
Review Comment:
Do you think it's the best way to fix it? I think this approach has a few drawbacks:
- As mentioned, lower efficiency of compression. Actually for some compression algorithms this may make it useless, if not harmful (if there is some significant amount of metadata that gets written on `flush`)
- It's somewhat strange that the result is different whether we call `getPos()` in the process of compressing the output, or we don't. This may later come as a big surprise for someone using `CompressibleFSDataOutputStream`.
- `Seek` operation in the corresponding `CompressibleFSDataInputStream` doesn't make sense for any position, only for some specific positions. Which is also not super bad, but confusing.
IMO, a better - albeit more involved - approach may be to change the semantics of `getPos()` for `CompressibleFSDataOutputStream`. This can represent the position in the **uncompressed** data.
Correspondingly, `seek` operation in `CompressibleFSDataInputStream` can be implemented having this semantics in mind: one can read the `compressingDelegate` until the desired position is reached.
This approach also doesn't have the problem of `seek` being valid for specific positions only, which is a nice side effect.
--
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
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "dmvk (via GitHub)" <gi...@apache.org>.
dmvk commented on code in PR #24079:
URL: https://github.com/apache/flink/pull/24079#discussion_r1450490276
##########
flink-runtime/src/main/java/org/apache/flink/runtime/state/CompressibleFSDataOutputStream.java:
##########
@@ -41,6 +41,10 @@ public CompressibleFSDataOutputStream(
@Override
public long getPos() throws IOException {
+ // Underlying compression involves buffering, so the only way to report correct position is
+ // to flush the underlying stream. This lowers the effectivity of compression, but there is
+ // no other way, since the position is often used as a split point.
+ flush();
Review Comment:
There is a general rule that compressed blocks can't be split. This is why most systems distinguish between per-file, per-block or per-record compression.
In this case we're using per-record compression. It could be optimized by some form of record batching / block compression, but that would need to happen further up the stack, which goes beyond a bug fix.
---
> It's somewhat strange that the result is different whether we call getPos() in the process of compressing the output, or we don't. This may later come as a big surprise for someone using CompressibleFSDataOutputStream.
Not really. The position corresponds to the uncompressed stream. If you haven't flushed any data into it .. the position won't move.
> IMO, a better - albeit more involved - approach may be to change the semantics of getPos() for CompressibleFSDataOutputStream. This can represent the position in the uncompressed data.
`getPos()` has a strictly defined contract on the interface level, we need to hold up to it. Chaging it for a single implementation violates Liskov Substitution Principle.
```
* <p>This method must report accurately report the current position of the stream. Various
* components of the high-availability and recovery logic rely on the accurate
```
---
As I said, this could clearly be optimized by switching from a per-record compression to block compression, but that goes beyond critical bugfix and should be addressed separately (I'll be happy to do a review if you have any ideas).
WDYT?
--
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
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "dmvk (via GitHub)" <gi...@apache.org>.
dmvk commented on PR #24079:
URL: https://github.com/apache/flink/pull/24079#issuecomment-1892561417
I've been able to pin down the issue to https://github.com/apache/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorStateRestoreOperation.java#L190
the problem is that the sorting breaks on empty state ... we should always restore in the order state was written
https://github.com/apache/flink/pull/24079/commits/3f4e553ba963753a3648cc96f5b9f9b91f08e976
cc @fredia @ruibinx since you've worked on https://github.com/apache/flink/pull/23938
--
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
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "dmvk (via GitHub)" <gi...@apache.org>.
dmvk commented on PR #24079:
URL: https://github.com/apache/flink/pull/24079#issuecomment-1893252213
@flinkbot run azure
--
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
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "isburmistrov (via GitHub)" <gi...@apache.org>.
isburmistrov commented on PR #24079:
URL: https://github.com/apache/flink/pull/24079#issuecomment-1892882299
> I'm actually starting to doubt whether the feature is even worth it for partitioned state. The compressed representation might actually end up being bigger than uncompressed one. To make it worth something, we should probably evolve partitioned state to use sparse index (still, it's a separate effort).
Yeah exactly. I wonder - is it hard to simply not do compression for partitioned state (but keep doing it for broadcast state)?
--
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
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "dmvk (via GitHub)" <gi...@apache.org>.
dmvk commented on PR #24079:
URL: https://github.com/apache/flink/pull/24079#issuecomment-1892858467
Fixed in https://github.com/apache/flink/pull/24079/commits/6bb32b114b76515757e602612dcf33668d031fef
I think this needs another review pass now. Can you please take another look @isburmistrov @dawidwys ?
--
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
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "dmvk (via GitHub)" <gi...@apache.org>.
dmvk commented on PR #24079:
URL: https://github.com/apache/flink/pull/24079#issuecomment-1892863335
Note that the latest fix (https://github.com/apache/flink/pull/24079#issuecomment-1892858467) is not BW compatible, but I don't think there is any harm in that since the snapshots were not restorable anyway.
--
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
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "dmvk (via GitHub)" <gi...@apache.org>.
dmvk commented on PR #24079:
URL: https://github.com/apache/flink/pull/24079#issuecomment-1894096708
> So that the order of our offset is no longer important, and compatibility does not need to be specially considered.
I don't think that needs to be considered at all, since the states are corrupted anyway, because the lack of flushes.
--
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
Re: [PR] [FLINK-34063][runtime] Always flush compression buffers, when retrieving stream position during OperatorState snapshot. [flink]
Posted by "dmvk (via GitHub)" <gi...@apache.org>.
dmvk merged PR #24079:
URL: https://github.com/apache/flink/pull/24079
--
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