FlinkSQL新增operatoer后基于savepoint重启报错怎么办？

原先sql任务是： CREATE TABLE A_source（...） CREATE TABLE B_sink （...） INSERT INTO B_sink SELECT 1 FROM A_source ; 我基于这个FlinkSQL任务生成了savepoint后，我重新修改为

CREATE TABLE A_source（...） CREATE TABLE B_sink （...） CREATE TABLE C_source（...） CREATE TABLE D_sink （...） INSERT INTO B_sink SELECT 1 FROM A_source ;

INSERT INTO C_sink SELECT 1 FROM D_source ; 并基于Savepoint提交，结果显示

Cannot map checkpoint/savepoint state for operator 2e9c6b0c053878cef673bbe7d94ab037 to the new program, because the operator is not available in the new program. If you want to allow to skip this, you can set the --allowNonRestoredState option on the CLI.

想请教一下底层是因为什么原因导致了opertor匹配不上？ *来自志愿者整理的flink邮件归档