[WIP][core] cut StoreSinkWriteState when write unaware bucket table

[wwj6591812]

Purpose

Now, when Flink bounded stream write to Paimon table. If the following conditions are met, checkpoint fail will occur, because paimon writer use flink UnionlistState. 1、If Source Operator is chained with Paimon writer Operater. 2、Some source Task has finished.

The exception is :
Caused by: org.apache.flink.runtime.checkpoint.FinishedTaskStateProvider$PartialFinishingNotSupportedByStateException: The vertex Source: SaroSource_ST_cf20240619200315_saro_mirror_mainsearch_result_table_c_d20240821200837_0[1136] -> Calc[1137] -> Map -> Writer(write-only) : mainse_result_stream_append_table_20240828 (id = 717c7b8afebbfb7137f6f0f99beb2a94) has used UnionListState, but part of its tasks has called operators' finish method.

After discuss, I have try cut StoreSinkWriteState when write unaware bucket table. But after test write to unaware bucket table, I failed. So, who can help me.

This is my code logical:

Linked issue: close #xxx

Tests

API and Format

Documentation