Hi, I have a problem about checkpoint and rescaling. I'm using Flink 1.16.0 with incremental checkpointing enabled and Flink operator 1.3.0. Few days ago I increased the job parallelism from 36 to 48 (max parallelism 4096) using

last-state

upgrade mode. After the job was restored, I noticed that the full checkpoint size drastically increased from 600GB to 900GB. To investigate this issue, I extract the internal RocksDB from one of the operator subtask state and use the

ldb

tool to dump the database. I find that there are a large number of deleted rows. The following is the command I use:

$ ldb --db=./db --column_family=name idump --hex | grep 'type:1' | wc -l
134529889
$ ldb --db=./db --column_family=name idump --hex | grep 'type:0' | wc -l
33779429

We can see that most of the rows are deleted. Next, I try to manually run compaction on the database and the size decreases from 16GB to 6.6GB. Why could rescaling create so many deleted rows in this case? Is there any configuration to alleviate the size increasing problem or it is just the expected behavior of rescaling?