[fhan688]

Purpose

related to: #4379

This PR introduces a read-only $partitions virtual table for querying partition metadata through standard Flink SQL.

Previously, partition IDs, names, and effective bucket counts could only be inspected through the sys.list_partition_infos procedure. A virtual table makes this metadata composable with regular SQL operations such as projection, filtering, aggregation, and joins.

For example:

SELECT partition_name, bucket_count
FROM my_db.orders$partitions
WHERE bucket_count > 2;

This is particularly useful after changing the table-level bucket.num, because existing partitions retain their original bucket counts while newly created partitions use the new value.

Brief change log

• Add the read-only $partitions virtual table with the following schema:
  • partition_id BIGINT NOT NULL
  • partition_name STRING NOT NULL
  • bucket_count INT NOT NULL
• Add catalog and table-factory routing for the new virtual table.
• Restrict $partitions to partitioned Fluss tables and reject mutation operations.
• Preserve $lake routing precedence for names such as table$lake$partitions.
• Add a bounded FLIP-27 source that loads partition metadata through Admin.listPartitionInfos.
• Load partition metadata in the source reader at execution time so cached SQL Gateway plans observe the latest partitions.
• Extract the singleton-split metadata source implementation shared by $buckets and $partitions.
• Deprecate sys.list_partition_infos while retaining it for backward compatibility.
• Document the new virtual table and update existing partition-bucketing documentation.

Predicate pushdown is not included in this initial implementation. Flink evaluates predicates after retrieving the complete partition metadata snapshot.

Tests

Added or extended tests covering:

• Partition metadata to RowData conversion.
• Lazy metadata loading and fresh metadata for each source execution.
• $partitions schema and catalog resolution.
• Table source creation in batch and streaming modes.
• Projection, filtering, and aggregation queries.
• Partitions with different effective bucket counts.
• Missing and non-partitioned base tables.
• Read-only table and sink behavior.
• $lake routing precedence.
• SQL Gateway plan-cache behavior, verifying that repeated execution of the same cached query observes newly created partitions.
• Regression coverage for the shared $buckets metadata source.

The relevant tests passed on Flink 1.18, 1.19, and 1.20. The final package build also passed Checkstyle, Spotless, and RAT validation.

API and Format

This change adds a new SQL-facing virtual table and deprecates the Java class backing sys.list_partition_infos.

The existing procedure remains available and its behavior is unchanged.

There are no changes to the storage format, network protocol, or persisted metadata format.

Documentation

The virtual-table documentation now describes:

• How to query $partitions.
• Its fixed schema and bounded-snapshot semantics.
• How to inspect effective per-partition bucket counts.
• The current lack of predicate pushdown.

The Flink procedure and bucketing documentation now recommend $partitions instead of sys.list_partition_infos.

[fhan688]

Could you please help review this PR when you have time? thanks! @swuferhong @platinumhamburg @Kaixuan-Duan

[platinumhamburg]

LGTM +1.