[Feature] Support bucket-level compaction for fixed-bucket tables

[lilei1128]

Search before asking

• I searched in the issues and found nothing similar.

Motivation

For non-partitioned tables with a large number of buckets, or partitioned tables where
each partition contains many buckets, a compaction job may need to scan and schedule all
buckets at once.

This can consume excessive memory during compaction planning and execution, and may
eventually cause OOM errors. In such cases, users currently have no way to split one large
compaction job into several smaller jobs.

We propose supporting compaction for selected buckets so users can compact a large table
in manageable batches.

Solution

Introduce an optional buckets parameter for compaction in both Flink and Spark.

The following interfaces will support this parameter:

• Flink compact action
• Flink compact procedure
• Spark compact procedure

The parameter accepts individual bucket IDs, comma-separated bucket IDs, and closed
bucket ranges:

5
1,3,5
0-99,200-299
0-99,150,200-299

Flink action example:

  compact \
    --warehouse hdfs:///path/to/warehouse \
    --database my_database \
    --table my_table \
    --compact_strategy full \
    --buckets 0-999

Flink and Spark procedure example:

  CALL sys.compact(
      `table` => 'my_database.my_table',
      compact_strategy => 'full',
      buckets => '0-999'
  );

For both Flink and Spark, bucket filtering should be applied before compaction tasks are
materialized. The filter should be pushed down into snapshot or manifest scanning so that
unselected buckets do not contribute to planning memory or task count.

Anything else?

No

Are you willing to submit a PR?

• I'm willing to submit a PR!