[core] make dynamic-bucket.initial-buckets immutable

[jacklong319]

Purpose

ISSUE

dynamic-bucket.initial-buckets affects assigner/bootstrap semantics for dynamic bucket tables.
For existing tables, changing this option may break Flink recovery/checkpoint for cross-partition upsert jobs, and may finally fail during bootstrap with duplicate-key related exceptions.
This PR marks dynamic-bucket.initial-buckets as immutable so that users get a clear validation error earlier instead of hitting runtime failures later.

Changes

• add @Immutable to CoreOptions.DYNAMIC_BUCKET_INITIAL_BUCKETS
• add tests to reject altering this option on existing tables

Tests

• manual test

API and Format

Documentation

Generative AI tooling

[jacklong319]

I also hit this issue in a real cross-partition upsert job.

Scenario:

• existing dynamic bucket table
• bucket = -1
• dynamic-bucket.initial-buckets was changed after the table had already contained data
• after restarting the Flink job, checkpoint failed during bootstrap of cross-partition-bucket-assigner

It seems that changing dynamic-bucket.initial-buckets on an existing table may break dynamic bucket assigner / bootstrap semantics, and finally cause duplicate-key related failure during bulk load.

The stack trace is like this:

java.io.IOException: Could not perform checkpoint 3 for operator cross-partition-bucket-assigner (23/128)#0. ... Caused by: java.lang.RuntimeException: Exception in bulkLoad, the most suspicious reason is that your data contains duplicates, please check your sink table. (The likelihood of duplication is that you used multiple jobs to write the same dynamic bucket table, it only supports single write) at org.apache.paimon.crosspartition.GlobalIndexAssigner.endBoostrapWithoutEmit(GlobalIndexAssigner.java:225) at org.apache.paimon.crosspartition.GlobalIndexAssigner.endBoostrap(GlobalIndexAssigner.java:200) at org.apache.paimon.flink.sink.index.GlobalIndexAssignerOperator.endBootstrap(GlobalIndexAssignerOperator.java:95) at org.apache.paimon.flink.sink.index.GlobalIndexAssignerOperator.prepareSnapshotPreBarrier(GlobalIndexAssignerOperator.java:85) ... Caused by: org.rocksdb.RocksDBException: Keys must be added in strict ascending order. at org.rocksdb.SstFileWriter.put(Native Method) at org.rocksdb.SstFileWriter.put(SstFileWriter.java:150) at org.apache.paimon.lookup.rocksdb.RocksDBBulkLoader.write(RocksDBBulkLoader.java:83) at org.apache.paimon.crosspartition.GlobalIndexAssigner.endBoostrapWithoutEmit(GlobalIndexAssigner.java:217)
So I think dynamic-bucket.initial-buckets should be treated as an immutable option for existing tables, instead of allowing the change and only failing later at runtime.

[jacklong319]

One more concern is the production impact of this issue.

For an already-used dynamic bucket table with a large amount of existing data, if dynamic-bucket.initial-buckets is changed by mistake, the Flink writing job may fail during checkpoint/bootstrap and become unable to run normally.

In practice, this is a serious online risk because:

• the table may already contain a large amount of production data
• the failure happens at runtime instead of being rejected at alter time
• the job may not be able to recover quickly
• rollback / repair / data cleanup may take a long time for large tables

So this is not only a usability issue, but also a stability and operability issue in production environments.

It would be much safer to reject changing dynamic-bucket.initial-buckets immediately for existing tables, instead of allowing the change and exposing users to a potentially major online incident later.