[FLINK-30113] Implement state compression for broadcast and regular operator states #21636

echauchot · 2023-01-10T09:34:17Z

What is the purpose of the change

Implement state compression for broadcast and regular operator states. Only the states themselves are compressed not the metadata.

Brief change log

Introduce FSData streams wrappers that add compression
Use these wrappers while writing state (in the default operator snapshot strategy) and while reading state (in the operator state restore operation)

Verifying this change

This change is already covered by existing tests, such as windowing ITCase tests ... in the flink-runtime module

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, ZooKeeper: yes checkpointing
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? yes
If yes, how is the feature documented? javadocs

R: @dawidwys

flinkbot · 2023-01-10T09:43:10Z

CI report:

83f0a2a Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

dawidwys · 2023-01-10T16:24:03Z

Thanks for the PR. I'll review it over the next few days.

MartijnVisser · 2023-01-11T13:28:52Z

@dawidwys Could you review this PR?

dawidwys

Hey @echauchot
Thanks for working on the issue.

Correct me if I am wrong, but in the current state you always apply the compression for the operator state.

First of all, imo, this needs to be configurable. I'd say it can be done with the same flag as for the keyed state backend (execution.checkpointing.snapshot-compression)

Moreover, right now it causes previously written checkpoints to be unreadable. (have you tried restoring from an old checkpoint? If not, could you add such a test?) as it always assumes the stream is compressed. The info if the state is compressed or not needs to be persisted somewhere. I don't have a good idea for that now. In the keyed state case it is done via KeyedBackendSerializationProxy#read/write.

Now I realise this might require changes to the checkpoint format, unless we figure out a smart place to put it, so we might need to create a small FLIP for that.

echauchot · 2023-01-16T09:36:15Z

Hey @echauchot Thanks for working on the issue.

Hey, thanks for the review !

Correct me if I am wrong, but in the current state you always apply the compression for the operator state.

No, see DefaultOperatorStateBackendBuilder I call AbstractStateBackend.getCompressionDecorator that reads execution.checkpointing.snapshot-compression conf parameter.

First of all, imo, this needs to be configurable. I'd say it can be done with the same flag as for the keyed state backend (execution.checkpointing.snapshot-compression)

See above, I've found we could reuse execution.checkpointing.snapshot-compression conf parameter. Tell me if you want it to be a different conf parameter.

Moreover, right now it causes previously written checkpoints to be unreadable. (have you tried restoring from an old checkpoint? If not, could you add such a test?) as it always assumes the stream is compressed.

Well, if the user disables compression to read uncompressed snapshots, then it works.
Though, I have not tested to restore old snapshots. I have just tested the state write/read roundtrip via existing tests. I'll add such a test

The info if the state is compressed or not needs to be persisted somewhere. I don't have a good idea for that now. In the keyed state case it is done via KeyedBackendSerializationProxy#read/write.

Yes I saw that with key state a boolean is written first when the state is written. I can do similar thing for operator state.

Now I realise this might require changes to the checkpoint format, unless we figure out a smart place to put it, so we might need to create a small FLIP for that.

Ok, adding the boolean will require to change the snapshot format indeed. This is why I have not added this boolean to avoid the format change. But, indeed, if a snapshot was compressed and the user then disables compression he could not read the previous compressed snapshot with my code. What about simply adding a compression boolean to operator state meta ? As a backward incompatible change, it requires a FLIP even if simple change, right ?

dawidwys · 2023-01-16T10:36:59Z

No, see DefaultOperatorStateBackendBuilder I call AbstractStateBackend.getCompressionDecorator that reads execution.checkpointing.snapshot-compression conf parameter.

Ah, sorry missed that :( That part looks good then.

Ok, adding the boolean will require to change the snapshot format indeed. This is why I have not added this boolean to avoid the format change. But, indeed, if a snapshot was compressed and the user then disables compression he could not read the previous compressed snapshot with my code. What about simply adding a compression boolean to operator state meta ? As a backward incompatible change, it requires a FLIP even if simple change, right ?

Yes, I do think it requires a FLIP as it is a substantial change, but I believe this change is unavoidable if we want to add operator state compression.

echauchot · 2023-01-16T13:15:51Z

No, see DefaultOperatorStateBackendBuilder I call AbstractStateBackend.getCompressionDecorator that reads execution.checkpointing.snapshot-compression conf parameter.

Ah, sorry missed that :( That part looks good then.

No problem.

Ok, adding the boolean will require to change the snapshot format indeed. This is why I have not added this boolean to avoid the format change. But, indeed, if a snapshot was compressed and the user then disables compression he could not read the previous compressed snapshot with my code. What about simply adding a compression boolean to operator state meta ? As a backward incompatible change, it requires a FLIP even if simple change, right ?

Yes, I do think it requires a FLIP as it is a substantial change, but I believe this change is unavoidable if we want to add operator state compression.

Ok, I'll write the FLIP and ping you on it for review, then when we agree on the architecture, I'll write the code and commit on this PR.

echauchot · 2023-01-16T15:43:57Z

Ok, I'll write the FLIP and ping you on it for review, then when we agree on the architecture, I'll write the code and commit on this PR.

@dawidwys I don't have the rights to add a FLIP to Flink confluence space. Can you give me the rights ?

MartijnVisser · 2023-01-17T08:35:16Z

Can you give me the rights ?

@echauchot You should have the rights now

echauchot · 2023-01-17T09:09:27Z

Can you give me the rights ?

@echauchot You should have the rights now

Thanks @MartijnVisser !

echauchot · 2023-01-17T13:53:41Z

Ok, I'll write the FLIP and ping you on it for review, then when we agree on the architecture, I'll write the code and commit on this PR.

@dawidwys I don't have the rights to add a FLIP to Flink confluence space. Can you give me the rights ?

@dawidwys here is the flip, can you please review it ?

echauchot · 2023-01-27T14:16:18Z

The vote on the FLIP has passed, but I'm going on vacation tonight for a week. I'll update this PR to reflect the FLIP when I get back

dawidwys · 2023-01-27T14:18:25Z

Absolutely @echauchot ! Thanks for the work so far!

echauchot · 2023-03-01T09:28:43Z

@dawidwys sorry for the delay, I had to finish this other PR after my vacations. Resuming the work on this PR ...

echauchot · 2023-03-01T14:10:37Z

@flinkbot run azure

echauchot · 2023-03-02T13:26:54Z

@flinkbot run azure

echauchot · 2023-03-02T15:31:52Z

Hi @dawidwys, my code is failing with versioning. I incremented the proxy version (as the metadata serde code has changed) and the snapshot version (as I added a boolean for compression). Can you explain how the proxy/snapshot versioning works because I see nothing in the docs.

dawidwys · 2023-03-06T10:56:55Z

What you did for versioning looks sane. I think the problem is with line

flink/flink-runtime/src/main/java/org/apache/flink/runtime/state/metainfo/StateMetaInfoSnapshotReadersWriters.java

Line 68 in a4a11a3

if (readVersion < CURRENT_STATE_META_INFO_SNAPSHOT_VERSION) {

which has not been implemented in a future proof way. The intent there was to throw an exception for readVersion < 6, not less than CURRENT_STATE_META_INFO_SNAPSHOT_VERSION. If we change the condition to readVersion < 6 it should work.

echauchot · 2023-03-06T14:17:38Z

@flinkbot run azure

echauchot · 2023-03-06T14:55:19Z

@flinkbot run azure

echauchot · 2023-03-07T08:34:06Z

What you did for versioning looks sane. I think the problem is with line

flink/flink-runtime/src/main/java/org/apache/flink/runtime/state/metainfo/StateMetaInfoSnapshotReadersWriters.java

Line 68 in a4a11a3

if (readVersion < CURRENT_STATE_META_INFO_SNAPSHOT_VERSION) {

which has not been implemented in a future proof way. The intent there was to throw an exception for readVersion < 6, not less than CURRENT_STATE_META_INFO_SNAPSHOT_VERSION. If we change the condition to readVersion < 6 it should work.

Hi @dawidwys, thanks for the prompt answer. I suspected something not future proof, thanks for the confirmation. It is now fixed. Tests pass but the CI fails in preparation of the E2E tests (since some days) with a http 404 error.

PTAL.

dawidwys

LGTM, let's wait until e2e tests are fixed though before we merge it.

echauchot · 2023-03-07T09:56:39Z

Thanks for the review Dawid !

MartijnVisser · 2023-03-07T13:09:35Z

LGTM, let's wait until e2e tests are fixed though before we merge it.

These E2E tests have been fixed for some time already, you probably need to rebase :) See https://issues.apache.org/jira/browse/FLINK-30972

…regular states

Add a new snapshot format version 6 compatible with the previous versions that adds a boolean for snapshot compression At snapshot time: depending on user configuration, write Snappy compressed regular and broadcast operator states (but not the snapshot meta info) and write the compression boolean accordingly At restoration time: depending on the read compression boolean, read the snapshots as compressed or not.

echauchot · 2023-03-07T13:26:52Z

LGTM, let's wait until e2e tests are fixed though before we merge it.

These E2E tests have been fixed for some time already, you probably need to rebase :) See https://issues.apache.org/jira/browse/FLINK-30972

Thanks @MartijnVisser, I missed this ticket. I just rebased. Let's wait for the green lights.

echauchot · 2023-03-07T13:27:34Z

@flinkbot run azure

echauchot · 2023-03-08T10:31:03Z

@dawidwys thanks for reviewing/merging.

echauchot changed the title ~~[FLINK-30113] Implement operator state compression for broadcast and regular states~~ [FLINK-30113] Implement state compression for broadcast and regular operator states Jan 10, 2023

echauchot force-pushed the FLINK-30113-compression-state-operator branch 2 times, most recently from d42f351 to e205374 Compare January 10, 2023 10:16

flinkbot added the component=Runtime/StateBackends label Jan 10, 2023

echauchot force-pushed the FLINK-30113-compression-state-operator branch from e205374 to 371763a Compare January 10, 2023 13:47

MartijnVisser requested a review from dawidwys January 11, 2023 13:28

dawidwys requested changes Jan 13, 2023

View reviewed changes

dawidwys self-assigned this Mar 6, 2023

echauchot force-pushed the FLINK-30113-compression-state-operator branch from a4a11a3 to be169cf Compare March 6, 2023 14:10

dawidwys approved these changes Mar 7, 2023

View reviewed changes

echauchot added 3 commits March 7, 2023 14:24

[FLINK-30113] Implement operator state compression for broadcast and …

3b80041

…regular states

[FLINK-30113] Increment both snapshot version and proxy version

83f0a2a

echauchot force-pushed the FLINK-30113-compression-state-operator branch from be169cf to 83f0a2a Compare March 7, 2023 13:24

dawidwys merged commit d04d727 into apache:master Mar 8, 2023

echauchot deleted the FLINK-30113-compression-state-operator branch March 17, 2023 14:51

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-30113] Implement state compression for broadcast and regular operator states #21636

[FLINK-30113] Implement state compression for broadcast and regular operator states #21636

echauchot commented Jan 10, 2023 •

edited

Loading

flinkbot commented Jan 10, 2023 •

edited

Loading

dawidwys commented Jan 10, 2023

MartijnVisser commented Jan 11, 2023

dawidwys left a comment

echauchot commented Jan 16, 2023 •

edited

Loading

dawidwys commented Jan 16, 2023 •

edited

Loading

echauchot commented Jan 16, 2023 •

edited

Loading

echauchot commented Jan 16, 2023

MartijnVisser commented Jan 17, 2023

echauchot commented Jan 17, 2023

echauchot commented Jan 17, 2023 •

edited

Loading

echauchot commented Jan 27, 2023

dawidwys commented Jan 27, 2023

echauchot commented Mar 1, 2023

echauchot commented Mar 1, 2023

echauchot commented Mar 2, 2023

echauchot commented Mar 2, 2023

dawidwys commented Mar 6, 2023

echauchot commented Mar 6, 2023

echauchot commented Mar 6, 2023

echauchot commented Mar 7, 2023

dawidwys left a comment

echauchot commented Mar 7, 2023

MartijnVisser commented Mar 7, 2023

echauchot commented Mar 7, 2023

echauchot commented Mar 7, 2023

echauchot commented Mar 8, 2023

[FLINK-30113] Implement state compression for broadcast and regular operator states #21636

[FLINK-30113] Implement state compression for broadcast and regular operator states #21636

Conversation

echauchot commented Jan 10, 2023 • edited Loading

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented Jan 10, 2023 • edited Loading

CI report:

dawidwys commented Jan 10, 2023

MartijnVisser commented Jan 11, 2023

dawidwys left a comment

Choose a reason for hiding this comment

echauchot commented Jan 16, 2023 • edited Loading

dawidwys commented Jan 16, 2023 • edited Loading

echauchot commented Jan 16, 2023 • edited Loading

echauchot commented Jan 16, 2023

MartijnVisser commented Jan 17, 2023

echauchot commented Jan 17, 2023

echauchot commented Jan 17, 2023 • edited Loading

echauchot commented Jan 27, 2023

dawidwys commented Jan 27, 2023

echauchot commented Mar 1, 2023

echauchot commented Mar 1, 2023

echauchot commented Mar 2, 2023

echauchot commented Mar 2, 2023

dawidwys commented Mar 6, 2023

echauchot commented Mar 6, 2023

echauchot commented Mar 6, 2023

echauchot commented Mar 7, 2023

dawidwys left a comment

Choose a reason for hiding this comment

echauchot commented Mar 7, 2023

MartijnVisser commented Mar 7, 2023

echauchot commented Mar 7, 2023

echauchot commented Mar 7, 2023

echauchot commented Mar 8, 2023

echauchot commented Jan 10, 2023 •

edited

Loading

flinkbot commented Jan 10, 2023 •

edited

Loading

echauchot commented Jan 16, 2023 •

edited

Loading

dawidwys commented Jan 16, 2023 •

edited

Loading

echauchot commented Jan 16, 2023 •

edited

Loading

echauchot commented Jan 17, 2023 •

edited

Loading