Hi Gabor, I applied your 1.20 patch, and I got some very good numbers from it... so for my 5GB savepoint, I made sure I skip all my code overhead to get the raw number, and I can read it in - HashMap : 4 minutes - RockDb with your patch: ~19 minutes - RockDb commenting out the remove(): 49 minutes
I am not sure these numbers make sense, and this is on a VM on my laptop, so not exactly a good performance testing environment, but the numbers I see are pretty good. I probably need more time to ensure we definitely see all the keys, but that looks very good. Hope this help JM #### using 1.20-SNAPSHOT + FLINK-37109_1.20.patch.txt: ( run 1/2) ``` 2025-02-12 08:22:59,282 INFO LifecycleStateExplorerTest - Starting now 2025-02-12 08:22:59,283 INFO c.i.a.l.s.LifecycleStateExplorer - Reading state from /tmp/mock_savepoint16385495638880357651. 2025-02-12 08:41:29,066 INFO LifecycleStateExplorerTest - Completed now 2025-02-12 08:41:29,066 INFO LifecycleStateExplorerTest - Duration: PT18M29.783388324S ms ``` #### using our flink 1.20 build, that removes the keysAndNamespaces.remove(); line ``` 2025-02-12 09:03:39,018 INFO LifecycleStateExplorerTest - Starting now 2025-02-12 09:03:39,024 INFO c.i.a.l.s.LifecycleStateExplorer - Reading state from /tmp/mock_savepoint10990862094634162213. 2025-02-12 09:52:51,811 INFO LifecycleStateExplorerTest - Completed now 2025-02-12 09:52:51,813 INFO LifecycleStateExplorerTest - Duration: PT49M12.788979538S ms ``` #### using 1.20-SNAPSHOT + FLINK-37109_1.20.patch.txt: (run 2/2) ``` 2025-02-12 10:12:26,453 INFO LifecycleStateExplorerTest - Starting now 2025-02-12 10:12:26,458 INFO c.i.a.l.s.LifecycleStateExplorer - Reading state from /tmp/mock_savepoint7967784813864743408. 2025-02-12 10:32:20,215 INFO LifecycleStateExplorerTest - Completed now 2025-02-12 10:32:20,215 INFO LifecycleStateExplorerTest - Duration: PT19M53.757274969S ms ``` #### using HashMapMemoryBackEnd (for reference) ``` 2025-02-12 10:39:03,618 INFO LifecycleStateExplorerTest - Starting now 2025-02-12 10:39:03,622 INFO c.i.a.l.s.LifecycleStateExplorer - Reading state from /tmp/mock_savepoint14340081990006677909. 2025-02-12 10:43:16,454 INFO LifecycleStateExplorerTest - Completed now 2025-02-12 10:43:16,457 INFO LifecycleStateExplorerTest - Duration: PT4M12.832810524S ms ``` On Tue, Feb 11, 2025 at 3:39 PM Gabor Somogyi <gabor.g.somo...@gmail.com> wrote: > Yeah, I think this is more like the 1.x and 2.x incompatibility. > I've just opened the PR agains 1.20 which you can cherry-pick here [1]. > > [1] https://github.com/apache/flink/pull/26145 > > BR, > G > > > On Tue, Feb 11, 2025 at 4:19 PM Jean-Marc Paulin <j...@paulin.co.uk> wrote: > >> Hi Gabor, >> >> So, a bit of progress, >> >> I managed to compile our stuff against your 2.1-SNAPSHOT (with a bit of >> chopping around deprecated/changed and removed APIs - that wasn't too bad), >> but that failed to read the state I was using before (that was generated >> with a Flink 1.20). This is the stack trace I get. I suspect this has more >> to do with state compatibility between 1.20 and 2.1... I was surprised the >> error is against a timer state. The end of the stack trace is below. >> >> I will try to apply your patch/change/PR to our 1.20 build, but it's not >> a simple git apply 😭😭😭. >> >> $ git apply --check ~/Downloads/26134.patch.txt >> error: patch failed: >> flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/KeyedStateInputFormat.java:226 >> error: >> flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/KeyedStateInputFormat.java: >> patch does not apply >> error: patch failed: >> flink-runtime/src/main/java/org/apache/flink/runtime/state/KeyedStateBackend.java:24 >> error: >> flink-runtime/src/main/java/org/apache/flink/runtime/state/KeyedStateBackend.java: >> patch does not apply >> error: >> flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionKeyedStateBackend.java: >> No such file or directory >> error: >> flink-runtime/src/test/java/org/apache/flink/streaming/runtime/tasks/TestStateBackend.java: >> No such file or directory >> error: patch failed: >> flink-state-backends/flink-statebackend-common/src/test/java/org/apache/flink/state/common/PeriodicMaterializationManagerTest.java:22 >> error: >> flink-state-backends/flink-statebackend-common/src/test/java/org/apache/flink/state/common/PeriodicMaterializationManagerTest.java: >> patch does not apply >> error: >> flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/sync/ForStSyncKeyedStateBackend.java: >> No such file or directory >> error: >> flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBKeyedStateBackend.java: >> No such file or directory >> error: >> flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/iterator/AbstractRocksStateKeysIterator.java: >> No such file or directory >> error: >> flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/iterator/RocksStateKeysIterator.java: >> No such file or directory >> >> >> stack trace if relevant... >> >> java.util.concurrent.CompletionException: java.lang.RuntimeException: >> Failed to fetch next result >> at >> com.ibm.aiops.lifecycle.stateexplorer.LifecycleStateExplorer.lambda$0(LifecycleStateExplorer.java:88) >> ~[classes/:?] >> at >> java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1768) >> ~[?:?] >> at >> java.base/java.util.concurrent.CompletableFuture$AsyncSupply.exec(CompletableFuture.java:1760) >> ~[?:?] >> at >> java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:373) >> ~[?:?] >> at >> java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1182) >> ~[?:?] >> at >> java.base/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1655) >> ~[?:?] >> at >> java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1622) >> ~[?:?] >> at >> java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:165) >> ~[?:?] >> Caused by: java.lang.RuntimeException: Failed to fetch next result >> ... many more >> Caused by: java.lang.RuntimeException: Failed to fetch next result >> ... many more >> Caused by: java.util.concurrent.ExecutionException: >> org.apache.flink.runtime.client.JobExecutionException: Job execution failed. >> ... 7 more >> Caused by: org.apache.flink.runtime.client.JobExecutionException: Job >> execution failed. >> ... many more >> Caused by: org.apache.flink.runtime.JobException: Recovery is suppressed >> by NoRestartBackoffTimeStrategy >> ... many more >> Caused by: java.io.IOException: Failed to restore timer state >> at >> org.apache.flink.state.api.input.KeyedStateInputFormat.open(KeyedStateInputFormat.java:193) >> ~[flink-state-processor-api-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.state.api.input.KeyedStateInputFormat.open(KeyedStateInputFormat.java:66) >> ~[flink-state-processor-api-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.streaming.api.functions.source.legacy.InputFormatSourceFunction.run(InputFormatSourceFunction.java:90) >> ~[flink-runtime-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:107) >> ~[flink-runtime-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:68) >> ~[flink-runtime-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:337) >> ~[flink-runtime-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> Caused by: java.lang.reflect.InvocationTargetException >> at >> java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance0(Native >> Method) ~[?:?] >> at >> java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:77) >> ~[?:?] >> at >> java.base/jdk.internal.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45) >> ~[?:?] >> at >> java.base/java.lang.reflect.Constructor.newInstanceWithCaller(Constructor.java:500) >> ~[?:?] >> at >> java.base/java.lang.reflect.Constructor.newInstance(Constructor.java:481) >> ~[?:?] >> at >> com.ibm.aiops.flink.state.StateManagerJsonNodeReaderFunction.open(StateManagerJsonNodeReaderFunction.java:52) >> ~[classes/:?] >> at >> org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:34) >> ~[flink-core-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.state.api.input.operator.StateReaderOperator.open(StateReaderOperator.java:106) >> ~[flink-state-processor-api-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.state.api.input.operator.KeyedStateReaderOperator.open(KeyedStateReaderOperator.java:66) >> ~[flink-state-processor-api-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.state.api.input.KeyedStateInputFormat.open(KeyedStateInputFormat.java:190) >> ~[flink-state-processor-api-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.state.api.input.KeyedStateInputFormat.open(KeyedStateInputFormat.java:66) >> ~[flink-state-processor-api-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.streaming.api.functions.source.legacy.InputFormatSourceFunction.run(InputFormatSourceFunction.java:90) >> ~[flink-runtime-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:107) >> ~[flink-runtime-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:68) >> ~[flink-runtime-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:337) >> ~[flink-runtime-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> Caused by: java.lang.RuntimeException: Error while getting state >> at >> org.apache.flink.runtime.state.DefaultKeyedStateStore.getMapState(DefaultKeyedStateStore.java:138) >> ~[flink-runtime-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.state.api.runtime.SavepointRuntimeContext.getMapState(SavepointRuntimeContext.java:232) >> ~[flink-state-processor-api-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> com.ibm.aiops.flink.state.RuntimeContextStateProvider.getState(RuntimeContextStateProvider.java:28) >> ~[classes/:?] >> at >> com.ibm.aiops.flink.state.StateManager.getState(StateManager.java:49) >> ~[classes/:?] >> at >> com.ibm.aiops.lifecycle.policy.execution.state.PolicyExecutionState.<init>(PolicyExecutionState.java:88) >> ~[classes/:?] >> at >> com.ibm.aiops.lifecycle.policy.execution.state.RequestExecutionStateManager.<init>(RequestExecutionStateManager.java:52) >> ~[classes/:?] >> at >> java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance0(Native >> Method) ~[?:?] >> at >> java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:77) >> ~[?:?] >> at >> java.base/jdk.internal.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45) >> ~[?:?] >> at >> java.base/java.lang.reflect.Constructor.newInstanceWithCaller(Constructor.java:500) >> ~[?:?] >> at >> java.base/java.lang.reflect.Constructor.newInstance(Constructor.java:481) >> ~[?:?] >> at >> com.ibm.aiops.flink.state.StateManagerJsonNodeReaderFunction.open(StateManagerJsonNodeReaderFunction.java:52) >> ~[classes/:?] >> at >> org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:34) >> ~[flink-core-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.state.api.input.operator.StateReaderOperator.open(StateReaderOperator.java:106) >> ~[flink-state-processor-api-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.state.api.input.operator.KeyedStateReaderOperator.open(KeyedStateReaderOperator.java:66) >> ~[flink-state-processor-api-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.state.api.input.KeyedStateInputFormat.open(KeyedStateInputFormat.java:190) >> ~[flink-state-processor-api-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.state.api.input.KeyedStateInputFormat.open(KeyedStateInputFormat.java:66) >> ~[flink-state-processor-api-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.streaming.api.functions.source.legacy.InputFormatSourceFunction.run(InputFormatSourceFunction.java:90) >> ~[flink-runtime-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:107) >> ~[flink-runtime-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:68) >> ~[flink-runtime-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:337) >> ~[flink-runtime-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> Caused by: org.apache.flink.util.StateMigrationException: The new state >> serializer >> (org.apache.flink.api.common.typeutils.base.MapSerializer@b45c430a) must >> not be incompatible with the old state serializer >> (org.apache.flink.api.common.typeutils.base.MapSerializer@edecbe44). >> at >> org.apache.flink.state.rocksdb.RocksDBKeyedStateBackend.updateRestoredStateMetaInfo(RocksDBKeyedStateBackend.java:858) >> ~[flink-statebackend-rocksdb-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.state.rocksdb.RocksDBKeyedStateBackend.tryRegisterKvStateInformation(RocksDBKeyedStateBackend.java:763) >> ~[flink-statebackend-rocksdb-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.state.rocksdb.RocksDBKeyedStateBackend.createOrUpdateInternalState(RocksDBKeyedStateBackend.java:1020) >> ~[flink-statebackend-rocksdb-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.state.rocksdb.RocksDBKeyedStateBackend.createOrUpdateInternalState(RocksDBKeyedStateBackend.java:1007) >> ~[flink-statebackend-rocksdb-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.runtime.state.KeyedStateFactory.createOrUpdateInternalState(KeyedStateFactory.java:47) >> ~[flink-runtime-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.runtime.state.ttl.TtlStateFactory.createStateAndWrapWithTtlIfEnabled(TtlStateFactory.java:70) >> ~[flink-runtime-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.runtime.state.AbstractKeyedStateBackend.getOrCreateKeyedState(AbstractKeyedStateBackend.java:384) >> ~[flink-runtime-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.runtime.state.AbstractKeyedStateBackend.getPartitionedState(AbstractKeyedStateBackend.java:435) >> ~[flink-runtime-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.runtime.state.DefaultKeyedStateStore.getPartitionedState(DefaultKeyedStateStore.java:150) >> ~[flink-runtime-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.runtime.state.DefaultKeyedStateStore.getMapState(DefaultKeyedStateStore.java:135) >> ~[flink-runtime-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.state.api.runtime.SavepointRuntimeContext.getMapState(SavepointRuntimeContext.java:232) >> ~[flink-state-processor-api-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> com.ibm.aiops.flink.state.RuntimeContextStateProvider.getState(RuntimeContextStateProvider.java:28) >> ~[classes/:?] >> at >> com.ibm.aiops.flink.state.StateManager.getState(StateManager.java:49) >> ~[classes/:?] >> at >> com.ibm.aiops.lifecycle.policy.execution.state.PolicyExecutionState.<init>(PolicyExecutionState.java:88) >> ~[classes/:?] >> at >> com.ibm.aiops.lifecycle.policy.execution.state.RequestExecutionStateManager.<init>(RequestExecutionStateManager.java:52) >> ~[classes/:?] >> at >> java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance0(Native >> Method) ~[?:?] >> at >> java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:77) >> ~[?:?] >> at >> java.base/jdk.internal.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45) >> ~[?:?] >> at >> java.base/java.lang.reflect.Constructor.newInstanceWithCaller(Constructor.java:500) >> ~[?:?] >> at >> java.base/java.lang.reflect.Constructor.newInstance(Constructor.java:481) >> ~[?:?] >> at >> com.ibm.aiops.flink.state.StateManagerJsonNodeReaderFunction.open(StateManagerJsonNodeReaderFunction.java:52) >> ~[classes/:?] >> at >> org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:34) >> ~[flink-core-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.state.api.input.operator.StateReaderOperator.open(StateReaderOperator.java:106) >> ~[flink-state-processor-api-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.state.api.input.operator.KeyedStateReaderOperator.open(KeyedStateReaderOperator.java:66) >> ~[flink-state-processor-api-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.state.api.input.KeyedStateInputFormat.open(KeyedStateInputFormat.java:190) >> ~[flink-state-processor-api-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.state.api.input.KeyedStateInputFormat.open(KeyedStateInputFormat.java:66) >> ~[flink-state-processor-api-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.streaming.api.functions.source.legacy.InputFormatSourceFunction.run(InputFormatSourceFunction.java:90) >> ~[flink-runtime-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:107) >> ~[flink-runtime-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:68) >> ~[flink-runtime-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> at >> org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:337) >> ~[flink-runtime-2.1-SNAPSHOT.jar:2.1-SNAPSHOT] >> >> >> >> On Tue, Feb 11, 2025 at 12:16 PM Jean-Marc Paulin <j...@paulin.co.uk> >> wrote: >> >>> Hi Gabor, >>> >>> Trying to but I struggle to compile my stuff against your Flink build... >>> tried to apply your PR as a patch on my 1.20 modified fork and that didn't >>> go well either. It will take time to untangle. >>> >>> Will keep you updated if I make progress >>> >>> JM >>> >>> >>> >>> On Mon, Feb 10, 2025 at 8:22 PM Gabor Somogyi <gabor.g.somo...@gmail.com> >>> wrote: >>> >>>> Hi Jean-Marc, >>>> >>>> FYI, I've just opened this [1] PR to address the issue in a clean way. >>>> May I ask you to test it on your side? >>>> >>>> [1] https://github.com/apache/flink/pull/26134 >>>> >>>> BR, >>>> G >>>> >>>> >>>> On Fri, Feb 7, 2025 at 6:14 PM Gabor Somogyi <gabor.g.somo...@gmail.com> >>>> wrote: >>>> >>>>> Just a little update on this. We've made our first POC with the >>>>> redesigned approach and the numbers are promising :) >>>>> It still requires huge efforts in development/correctness/performance >>>>> perspective but seems like we have something in the pocket. >>>>> >>>>> Test data: 256Mb state file with a single operator and 2 value states >>>>> - Old execution time: 25M27.126737S >>>>> - New execution time: 1M19.602042S >>>>> In short: ~95% performance gain. >>>>> >>>>> G >>>>> >>>>> >>>>> On Thu, Feb 6, 2025 at 9:06 AM Gabor Somogyi < >>>>> gabor.g.somo...@gmail.com> wrote: >>>>> >>>>>> In short, when you don't care about >>>>>> multiple KeyedStateReaderFunction.readKey calls then you're on the safe >>>>>> side. >>>>>> >>>>>> G >>>>>> >>>>>> On Wed, Feb 5, 2025 at 6:27 PM Jean-Marc Paulin <j...@paulin.co.uk> >>>>>> wrote: >>>>>> >>>>>>> I am still hoping that I am still good. I just read the savepoint to >>>>>>> extract information (parallelism 1, and only 1 task manager) . I also >>>>>>> know >>>>>>> it has been created by a job using a HashMap backend. And I do not care >>>>>>> about duplicates. >>>>>>> >>>>>>> I should still be good, right? from what I saw I never read any >>>>>>> duplicate keys. >>>>>>> >>>>>>> Thanks >>>>>>> >>>>>>> JM >>>>>>> >>>>>>> >>>>>>> On Wed, Feb 5, 2025 at 4:58 PM Gabor Somogyi < >>>>>>> gabor.g.somo...@gmail.com> wrote: >>>>>>> >>>>>>>> Hi Guys, >>>>>>>> >>>>>>>> We've just had an in-depth analysis and we think that removing that >>>>>>>> particular line causes correctness issues under some circumstances. >>>>>>>> >>>>>>>> Namely key duplicates can happen when multiple column families are >>>>>>>> processed at the same time. Not need to mention that it would cause >>>>>>>> multiple >>>>>>>> `readKey` calls which ends up in just wrong calculation logic (for >>>>>>>> example in simple sum calculation). >>>>>>>> >>>>>>>> We've a vision how this can be solved in a clean way but it will >>>>>>>> take some time. >>>>>>>> >>>>>>>> > Are there any plans on a migration guide or something for users >>>>>>>> to adapt their QS observers (beyond the current docs)? >>>>>>>> >>>>>>>> The gap between the two approaches are quite huge and considering >>>>>>>> the actual bugs and improvement possibilities in the state processor >>>>>>>> API >>>>>>>> I would say this can come later on at least on my plate. When you >>>>>>>> see the gaps and you know how to fill them feel free to contribute and >>>>>>>> we can shepherd the PRs. >>>>>>>> >>>>>>>> G >>>>>>>> >>>>>>>> >>>>>>>> On Wed, Feb 5, 2025 at 10:57 AM Salva Alcántara < >>>>>>>> salcantara...@gmail.com> wrote: >>>>>>>> >>>>>>>>> Thanks both for your work on this! >>>>>>>>> >>>>>>>>> On a related note, since Queryable State (QS) is going away soon, >>>>>>>>> streamlining the State Processor API as much as possible makes a lot >>>>>>>>> of >>>>>>>>> sense. >>>>>>>>> >>>>>>>>> Are there any plans on a migration guide or something for users to >>>>>>>>> adapt their QS observers (beyond the current docs)? >>>>>>>>> (State-)Observability-wise Flink has some room for improvement I >>>>>>>>> would say. >>>>>>>>> >>>>>>>>> Regards, >>>>>>>>> >>>>>>>>> Salva >>>>>>>>> >>>>>>>>> >>>>>>>>> On Wed, Feb 5, 2025 at 9:36 AM Gabor Somogyi < >>>>>>>>> gabor.g.somo...@gmail.com> wrote: >>>>>>>>> >>>>>>>>>> Hi Jean-Marc, >>>>>>>>>> >>>>>>>>>> Thanks for your time investment and to share the numbers, it's >>>>>>>>>> super helpful. >>>>>>>>>> Ping me any time when you have further info to share. >>>>>>>>>> >>>>>>>>>> About the numbers: 48 minutes for 6Gb is not good but not >>>>>>>>>> terrible. >>>>>>>>>> I've seen petabyte scale states so I'm pretty sure we need to go >>>>>>>>>> beyond... >>>>>>>>>> >>>>>>>>>> Since we measure similar numbers with the unpatched Flink plus >>>>>>>>>> this has been reported this by several users, >>>>>>>>>> we must make changes in this area. It's still a question whether >>>>>>>>>> the tested patch is the right approach >>>>>>>>>> but at least we've touched the root cause. >>>>>>>>>> >>>>>>>>>> The next step on my side is to have a deep dive and understand >>>>>>>>>> all the aspects why remove is there, >>>>>>>>>> how the remove elimination would effect existing use-cases and >>>>>>>>>> consider all other possibilities. >>>>>>>>>> >>>>>>>>>> BR, >>>>>>>>>> G >>>>>>>>>> >>>>>>>>>> >>>>>>>>>> On Wed, Feb 5, 2025 at 9:05 AM Jean-Marc Paulin <j...@paulin.co.uk> >>>>>>>>>> wrote: >>>>>>>>>> >>>>>>>>>>> Hi Gabor, >>>>>>>>>>> >>>>>>>>>>> I finally got to run that change through. I have a 6Gb savepoint >>>>>>>>>>> I read and parse for reference. >>>>>>>>>>> - HashMap reads it in 14 minutes (but requires 10 Gb of RAM) >>>>>>>>>>> - RockDb with the patch reads it in 48 minutes (and requires >>>>>>>>>>> less than 2Gb) >>>>>>>>>>> - RockDb without the patch wasn't even halfway through after 12 >>>>>>>>>>> hours.... (I gave up) >>>>>>>>>>> >>>>>>>>>>> I don't think I have any duplicates because the application that >>>>>>>>>>> generates the savepoint is using HashMap, so my scenario may not be >>>>>>>>>>> representative. I am using IBM Seremu Java 17 (openJ9-0.46). >>>>>>>>>>> >>>>>>>>>>> That was run on a VM on my laptop, so not exactly a >>>>>>>>>>> controlled environment. but I think it's conclusive enough. I will >>>>>>>>>>> need to >>>>>>>>>>> run further tests but I think we will patch our Flink. using a >>>>>>>>>>> system >>>>>>>>>>> property to configure it. >>>>>>>>>>> >>>>>>>>>>> Hope this help >>>>>>>>>>> >>>>>>>>>>> JM >>>>>>>>>>> >>>>>>>>>>> On Tue, Feb 4, 2025 at 4:01 PM Gabor Somogyi < >>>>>>>>>>> gabor.g.somo...@gmail.com> wrote: >>>>>>>>>>> >>>>>>>>>>>> Just to give an update. I've applied the mentioned patch and >>>>>>>>>>>> the execution time drastically decreased (the gain is 98.9%): >>>>>>>>>>>> >>>>>>>>>>>> 2025-02-04 16:52:54,448 INFO o.a.f.e.s.r.FlinkTestStateReader >>>>>>>>>>>> [] - Execution time: PT14.690426S >>>>>>>>>>>> >>>>>>>>>>>> I need to double check what that would mean to correctness and >>>>>>>>>>>> all other aspects. >>>>>>>>>>>> >>>>>>>>>>>> G >>>>>>>>>>>> >>>>>>>>>>>> >>>>>>>>>>>> On Tue, Feb 4, 2025 at 2:26 PM Gabor Somogyi < >>>>>>>>>>>> gabor.g.somo...@gmail.com> wrote: >>>>>>>>>>>> >>>>>>>>>>>>> Please report back on how the patch behaves including any side >>>>>>>>>>>>> effects. >>>>>>>>>>>>> >>>>>>>>>>>>> Now I'm in testing the state reading with processor API vs the >>>>>>>>>>>>> mentioned job where we control the keys. >>>>>>>>>>>>> The difference is extreme, especially because the numbers are >>>>>>>>>>>>> coming from reading ~40Mb state file😅 >>>>>>>>>>>>> >>>>>>>>>>>>> 2025-02-04 13:21:53,580 INFO o.a.f.e.s.r.FlinkTestStateReader >>>>>>>>>>>>> [] - Execution time: PT22M24.612954S >>>>>>>>>>>>> ... >>>>>>>>>>>>> 2025-02-04 13:39:14,704 INFO o.a.f.e.s.r.FlinkTestStateReaderJob >>>>>>>>>>>>> [] - Execution time: PT6.930659S >>>>>>>>>>>>> >>>>>>>>>>>>> Don't need to mention that the bigger is the processor API. >>>>>>>>>>>>> >>>>>>>>>>>>> G >>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>>> On Tue, Feb 4, 2025 at 1:40 PM Jean-Marc Paulin < >>>>>>>>>>>>> j...@paulin.co.uk> wrote: >>>>>>>>>>>>> >>>>>>>>>>>>>> That's a good idea, Sadly I have no control over the keys.... >>>>>>>>>>>>>> >>>>>>>>>>>>>> I was going to patch Flink with the suggestion in FLINK-37109 >>>>>>>>>>>>>> <https://issues.apache.org/jira/browse/FLINK-37109> first to >>>>>>>>>>>>>> see how that goes. If that brings RockDb performance in an >>>>>>>>>>>>>> acceptable range >>>>>>>>>>>>>> for us we might go that way. I really like the light memory >>>>>>>>>>>>>> consumption of >>>>>>>>>>>>>> RockDb for that kind of side job. >>>>>>>>>>>>>> >>>>>>>>>>>>>> Thanks >>>>>>>>>>>>>> >>>>>>>>>>>>>> JM >>>>>>>>>>>>>> >>>>>>>>>>>>>> On Tue, Feb 4, 2025 at 12:23 PM Gabor Somogyi < >>>>>>>>>>>>>> gabor.g.somo...@gmail.com> wrote: >>>>>>>>>>>>>> >>>>>>>>>>>>>>> What I could imagine is to create a normal Flink job, >>>>>>>>>>>>>>> use execution.state-recovery.path=/path/to/savepoint >>>>>>>>>>>>>>> set the operator UID on a custom written operator, which >>>>>>>>>>>>>>> opens the state info for you. >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> The only drawback is that you must know the keyBy range... >>>>>>>>>>>>>>> this can be problematic but if you can do it it's a win :) >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> G >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> On Tue, Feb 4, 2025 at 12:16 PM Jean-Marc Paulin < >>>>>>>>>>>>>>> j...@paulin.co.uk> wrote: >>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> Hi Gabor, >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> I thought so. I was hoping for a way to read the savepoint >>>>>>>>>>>>>>>> in pages, instead of as a single blob up front which I think >>>>>>>>>>>>>>>> is what the >>>>>>>>>>>>>>>> hashmap does... we just want to be called for each entry and >>>>>>>>>>>>>>>> extract the >>>>>>>>>>>>>>>> bit we want in that scenario. >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> Never mind >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> Thank you for the insight. Saves me a lot of hunting for >>>>>>>>>>>>>>>> nothing. >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> JM >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> On Tue, Feb 4, 2025 at 10:45 AM Gabor Somogyi < >>>>>>>>>>>>>>>> gabor.g.somo...@gmail.com> wrote: >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> Hi Jean-Marc, >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> We've already realized that the RocksDB approach is not >>>>>>>>>>>>>>>>> reaching the performance criteria which it should be. There >>>>>>>>>>>>>>>>> is an open >>>>>>>>>>>>>>>>> issue for it [1]. >>>>>>>>>>>>>>>>> The hashmap based approach was and is always expecting >>>>>>>>>>>>>>>>> more memory. So if the memory footprint is a hard requirement >>>>>>>>>>>>>>>>> then RocksDB >>>>>>>>>>>>>>>>> is the only way now. >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> Bad to say but I can't suggest any nifty trick to make it >>>>>>>>>>>>>>>>> better. All I can promise that I'm now measuring performance >>>>>>>>>>>>>>>>> of the RocksDB >>>>>>>>>>>>>>>>> approach >>>>>>>>>>>>>>>>> and intended to eliminate the slowness. Since we don't >>>>>>>>>>>>>>>>> know what exactly causes the slowness the new Frocksdb-8.10.0 >>>>>>>>>>>>>>>>> can be also >>>>>>>>>>>>>>>>> an imrpvement. >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> All in all it will take some time to sort this out. >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> [1] https://issues.apache.org/jira/browse/FLINK-37109 >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> BR, >>>>>>>>>>>>>>>>> G >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> On Tue, Feb 4, 2025 at 10:29 AM Jean-Marc Paulin < >>>>>>>>>>>>>>>>> jm.pau...@gmail.com> wrote: >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> What would be the best approach to read a savepoint and >>>>>>>>>>>>>>>>>> minimise the memory consumption. We just need to transform >>>>>>>>>>>>>>>>>> it into >>>>>>>>>>>>>>>>>> something else for investigation. >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> Our flink 1.20 streaming job is using HashMap backend, >>>>>>>>>>>>>>>>>> and is spread over 6 task slots in 6 pods (under k8s). >>>>>>>>>>>>>>>>>> Savepoints are saved >>>>>>>>>>>>>>>>>> on S3. A savepoint can be 4-5Gb or more. >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> The reader is more basic, using a Local Execution >>>>>>>>>>>>>>>>>> EnvironmentThis is essentially what we are doing: >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> StreamExecutionEnvironment env = >>>>>>>>>>>>>>>>>> LocalStreamEnvironment.getExecutionEnvironment(); >>>>>>>>>>>>>>>>>> env.setParallelism(1); >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> SavepointReader savepoint = >>>>>>>>>>>>>>>>>> SavepointReader.read(env, savepointLocation, new >>>>>>>>>>>>>>>>>> HashMapStateBackend()); >>>>>>>>>>>>>>>>>> // SavepointReader.read(env, savepointLocation, new >>>>>>>>>>>>>>>>>> EmbeddedRocksDBStateBackend()); // Too slow >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> DataStream<MainOperatorStateJsonNodeReaderFunction.KeyedStateItem> >>>>>>>>>>>>>>>>>> mainOperatorState = >>>>>>>>>>>>>>>>>> savepoint.readKeyedState( >>>>>>>>>>>>>>>>>> MAIN_OPERATOR, >>>>>>>>>>>>>>>>>> new >>>>>>>>>>>>>>>>>> StateManagerJsonNodeReaderFunction<>(StateManager.class)); >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> CloseableIterator<MainOperatorStateJsonNodeReaderFunction.KeyedStateItem> >>>>>>>>>>>>>>>>>> stateReader = mainOperatorState.executeAndCollect() >>>>>>>>>>>>>>>>>> stateReader.forEachRemaining( record -> { ... >>>>>>>>>>>>>>>>>> /// extract what we need here >>>>>>>>>>>>>>>>>> } >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> We tried two approaches: >>>>>>>>>>>>>>>>>> - One is to read the savepoint with a rockDb backend. >>>>>>>>>>>>>>>>>> That works and is low on memory usage, but is very very >>>>>>>>>>>>>>>>>> slow. We noticed >>>>>>>>>>>>>>>>>> the iterator is available very early on, but it is slow... >>>>>>>>>>>>>>>>>> - The other is to read the savepoint with a HashMap >>>>>>>>>>>>>>>>>> backend. That is very fast, as expected. However the >>>>>>>>>>>>>>>>>> iterator apparently >>>>>>>>>>>>>>>>>> only returns once the whole savepoint has been loaded in the >>>>>>>>>>>>>>>>>> HashMap, so >>>>>>>>>>>>>>>>>> heavy memory consumption. >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> Is there a better way to do that? or a way to tune it so >>>>>>>>>>>>>>>>>> that it does not consume all the memory ? or maybe reading >>>>>>>>>>>>>>>>>> it in parts... >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> Thanks >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> JM >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>
