Hey all, I just created a new JIRA improvement issue for this: https://issues.apache.org/jira/browse/FLINK-41424. After searching I see there may be an older outstanding improvmenet: https://issues.apache.org/jira/browse/FLINK-10121. If both these issues are truly the same I’m happy to delete mine.
Since it does look like this would be net new work would the process now be to create a FLIP to drive this forward or would something like this be small enough to be handled as part of a JIRA issue? Thanks, Francis > On Oct 22, 2026, at 12:21, Francis Altomare <[email protected]> > wrote: > > Hi all, > > We run a Flink 2.2.1 job on ForSt with the async state API, incremental > checkpoints and CLAIM restores. Over time we removed three ValueState > descriptors from a single keyed operator, but their column families never > went away — the data continued to be carried over into all future > checkpoints. In our specific case this mean that around 50% of our checkpoint > file size was unmapped state. > > None of the options we found seemed to fix this: > - allowNonRestoredState only works per operator. > - The State Processor API means stopping the job and rewriting a savepoint, > and it doesn't support ForSt. > - applyToAllKeys isn't available on the async ForSt backend. > - Adding a TTL to the existing state needs a state migration, which async > ForSt doesn't support yet > > What we ended up doing is re-registering the removed descriptors in open() > and running Flink's own rescale code on just those column families. In order > to make this work we needed to get a handle on the DB insrance requiring > reflective reads and a bit of hacky code: > > val states = deadStateNames.map { > runtimeContext.getState(ValueStateDescriptor(it, > BYTE_PRIMITIVE_ARRAY_TYPE_INFO)) > }.filterIsInstance<ForStValueState<*, *, *>>() > > // AsyncExecutionController#switchContextListener is the keyed backend > (package-private) > val backend = readField( > states.first().stateRequestHandler, > AsyncExecutionController::class.java, > "switchContextListener", > ) as ForStKeyedStateBackend<*> > > // ForStKeyedStateBackend#db is protected > val db = readField(backend, ForStKeyedStateBackend::class.java, "db") as > RocksDB > > With these handles we could use the existing > ForStIncrementalCheckpointUtils.clipDBWithKeyGroupRange: > ForStIncrementalCheckpointUtils.clipDBWithKeyGroupRange( > db, > states.map { it.columnFamilyHandle }, > KeyGroupRange.EMPTY_KEY_GROUP_RANGE, > backend.keyGroupRange, > > computeRequiredBytesInKeyGroupPrefix(runtimeContext.taskInfo.maxNumberOfParallelSubtasks), > true, > ) > > Related to this, I have a few questions: > 1. Is there a supported way to do this that we missed? > 2. Are there pitfalls with deleteFilesInRanges on restored (NOT_OWNED) files > beyond FLINK-37686 <https://github.com/apache/flink/pull/28801>? > > Thanks a lot, > > Francis
