[flink] Support compact_chain_table procedure

[juntaozhang]

Purpose

see #8503

Tests

• CompactChainTableActionITCase.java
• CompactChainTableProcedureITCase.java

[JingsongLi]

compact_chain_table(..., partial partition, overwrite=true) may cause data loss. Using the partial partitionSpec in sinkBuilder.overwrite(partitionSpec) will delete all groups under the specified partial partition in the snapshot branch. However, the source side's skipPreloadTargetSnapshot only generates output from target chain partitions present in the delta. If a snapshot group already exists but has no corresponding delta in the target chain partition, it will be overwritten and deleted without being rewritten. The author is advised to add a scenario: partitioned by (region, dt, hr), chain keys (dt, hr), with a snapshot containing region=FR, dt=..., hr=..., and a delta only including region=CN, dt=..., hr=.... After executing compact_chain_table('dt=..., hr=...', true), the FR data should not be lost.

[juntaozhang]

compact_chain_table(..., partial partition, overwrite=true) may cause data loss. Using the partial partitionSpec in sinkBuilder.overwrite(partitionSpec) will delete all groups under the specified partial partition in the snapshot branch. However, the source side's skipPreloadTargetSnapshot only generates output from target chain partitions present in the delta. If a snapshot group already exists but has no corresponding delta in the target chain partition, it will be overwritten and deleted without being rewritten. The author is advised to add a scenario: partitioned by (region, dt, hr), chain keys (dt, hr), with a snapshot containing region=FR, dt=..., hr=..., and a delta only including region=CN, dt=..., hr=.... After executing compact_chain_table('dt=..., hr=...', true), the FR data should not be lost.

Hi @JingsongLi

Thank you for pointing this out.
I updated CompactChainTableAction to always enable dynamic-partition-overwrite internally when compacting, so only full partitions present in the compacted DataFrame are overwritten and other groups are preserved regardless of the table-level setting.

I also verified that the same issue exists in the Spark compact_chain_table implementation. I have submitted a separate PR(#8529) to fix it there as well.

[JingsongLi]

+1