[Bug] After upgrading Paimon from 0.7 to 0.8, Flink state is incompatible

[liming30]

Search before asking

• I searched in the issues and found nothing similar.

Paimon version

0.8

Compute Engine

flink-1.17

Minimal reproduce step

1. Use version 0.7 of paimon-flink-1.17 to write to the paimon table and make a checkpoint;
2. after upgrading paimon-flink-1.17 to version 0.8, restore and restart from the previous checkpoint.

What doesn't meet your expectations?

Commit Operator throws exception:

java.lang.ArrayIndexOutOfBoundsException: Index 1 out of bounds for length 1
	at org.apache.paimon.memory.MemorySegmentUtils.getIntSlowly(MemorySegmentUtils.java:695)
	at org.apache.paimon.memory.MemorySegmentUtils.getIntMultiSegments(MemorySegmentUtils.java:685)
	at org.apache.paimon.memory.MemorySegmentUtils.getInt(MemorySegmentUtils.java:673)
	at org.apache.paimon.data.BinaryArray.pointTo(BinaryArray.java:127)
	at org.apache.paimon.memory.MemorySegmentUtils.readArrayData(MemorySegmentUtils.java:1140)
	at org.apache.paimon.data.BinaryRow.getArray(BinaryRow.java:337)
	at org.apache.paimon.index.IndexFileMetaSerializer.fromRow(IndexFileMetaSerializer.java:58)
	at org.apache.paimon.index.IndexFileMetaSerializer.fromRow(IndexFileMetaSerializer.java:33)
	at org.apache.paimon.utils.ObjectSerializer.deserialize(ObjectSerializer.java:81)
	at org.apache.paimon.utils.ObjectSerializer.deserializeList(ObjectSerializer.java:104)
	at org.apache.paimon.table.sink.CommitMessageSerializer.deserialize(CommitMessageSerializer.java:127)
	at org.apache.paimon.table.sink.CommitMessageSerializer.deserializeList(CommitMessageSerializer.java:98)
	at org.apache.paimon.manifest.ManifestCommittableSerializer.deserialize(ManifestCommittableSerializer.java:94)
	at org.apache.paimon.manifest.ManifestCommittableSerializer.deserialize(ManifestCommittableSerializer.java:34)
	at org.apache.paimon.flink.VersionedSerializerWrapper.deserialize(VersionedSerializerWrapper.java:48)
	at org.apache.flink.core.io.SimpleVersionedSerialization.readVersionAndDeSerialize(SimpleVersionedSerialization.java:227)
	at org.apache.flink.streaming.api.operators.util.SimpleVersionedListState$DeserializingIterator.next(SimpleVersionedListState.java:138)
	at java.base/java.lang.Iterable.forEach(Iterable.java:74)
	at org.apache.paimon.flink.sink.RestoreAndFailCommittableStateManager.initializeState(RestoreAndFailCommittableStateManager.java:75)
	at org.apache.paimon.flink.sink.CommitterOperator.initializeState(CommitterOperator.java:123)
	at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.initializeOperatorState(StreamOperatorStateHandler.java:122)
	at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:292)
	at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:282)
	at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.initializeStateAndOpenOperators(RegularOperatorChain.java:107)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:742)
	at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreInternal(StreamTask.java:717)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:682)
	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:969)
	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:938)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:762)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:572)
	at java.base/java.lang.Thread.run(Thread.java:829)

Anything else?

From version 0.8, CommitMessageSerializer has added serialization/deserialization of deletedFiles, but it is not compatible with the old version.

Are you willing to submit a PR?

• I'm willing to submit a PR!