feat: Support bucket state in filesystem writer

[KevinyhZou]

Purpose

为 FileSystem Writer 补齐与 Flink StreamingFileSink / Bucket 对齐的 checkpoint snapshot & restore 能力，使 failover / 重启后能恢复 writer 的 bucket 状态（尤其是 partCounter 与未提交 pending 文件），避免 part 文件号回退导致覆盖已提交文件，并保证 barrier 之后的写出不会被错误提交到更早的 checkpoint。

Changes

Connector / Operator API

• 在 DataSink 上扩展 snapshot() / restoreState() 接口，用于按 checkpoint 封印写出并恢复 sink 状态。
• TableWriter 将 stateful 算子的 snapshotState / restoreState / commit 转发到底层 DataSink。

FileSystemDataSink

• 实现 snapshot(checkpointId)：关闭当前 writers，把 pending writer 归入对应 checkpoint，并序列化 bucket state（JSON）。
• 实现 restoreState(checkpointRecords)：从 checkpoint 记录恢复：
  • 全局 maxPartCounter（对齐 Flink partCounterState）
  • 未完成的 pendingFiles
  • 兼容旧版按 bucket 存 partCounter 的 state 格式（restore 时取 max）
• FsFileNameGenerator：仅 checkpoint 全局 maxPartCounter；各 bucket 内存中的 partCounter 从该全局值初始化，防止 restore 后覆盖已提交 part 文件。
• commit(checkpointId)：只提交属于该（及更早）checkpoint 的 pending 文件；barrier 后新写入不进入已 snapshot 的 checkpoint。

Tests

• 补充 FileSystemConnectorTest 覆盖：
  • snapshot/commit 与 barrier 后写入隔离
  • restore 后 partCounter 延续
  • 旧版 per-bucket partCounter state 的兼容恢复

Notes

Checkpoint state 格式与 Flink StreamingFileSink 对齐：checkpoint 全局 partCounter + pendingFiles；partition commit policy（如 _SUCCESS / metastore）仍由 Flink 侧基于 commit() 返回的路径处理。