feat(datafusion): support REST format table partition commands

[sundapeng]

Purpose

A Format Table loaded from REST Catalog can have its partitions managed by the catalog instead of discovered from the directory layout. DataFusion had no way to administer that partition set, and no way to read it: SHOW PARTITIONS, ALTER TABLE ... ADD/DROP PARTITION and MSCK REPAIR TABLE were unavailable, and a scan discovered partitions from storage even when the catalog was the authority.

This change adds those commands and makes the catalog registration the partition set a scan reads.

Brief change log

• Add REST Catalog APIs for paginated partition listing and create/drop registration, with retry-safe 1000-entry batching for idempotent requests.
• Load eligible internal Format Tables with immutable REST partition metadata and scan only registered partitions.
• Add DataFusion support for:
  • SHOW PARTITIONS table [PARTITION (...)]
  • ALTER TABLE table ADD [IF NOT EXISTS] PARTITION (...)
  • ALTER TABLE table DROP [IF EXISTS] PARTITION (...)
  • MSCK REPAIR TABLE table [{ADD|DROP|SYNC} PARTITIONS]
• DROP PARTITION follows Java Format Table semantics: one statement may carry several specifications, and a specification that fixes only some of the partition keys expands to every registered partition it matches. The fixed keys need not be a leading prefix, so hh = '10' alone works on a (dt, hh) table. One catalog listing serves the whole statement.
• A catalog-managed scan pushes the leading equality prefix of its filter down as partitionNamePattern with maxResults=1000, instead of downloading every registration and pruning locally. The pattern is a hint the catalog may ignore, so the local match stays as the backstop.
• Partition boolean values accept every spelling Java accepts (t/true/y/yes/1 and their negatives, case-insensitive), so a partition another engine registered as TRUE is readable here.
• Normalize typed partition literals, default partition values, and Java-compatible partition paths.
• Keep repair metadata-only and fail closed before destructive DROP/SYNC reconciliation.

Tests

• crates/paimon/tests/format_partition_test.rs, rest_api_test.rs, rest_catalog_test.rs
• crates/integrations/datafusion/tests/rest_format_partition_sql.rs, including:
  • a partial specification expanding to every matching partition, a non-leading one, and several specifications in one statement
  • a complete specification that matches nothing raising unless IF EXISTS, a partial one that matches nothing being a no-op, and a failing specification leaving the whole statement unapplied
  • the partitionNamePattern a scan pushes for each predicate shape, asserted on the request the server received
• Unit tests for the pattern builder and for boolean partition values in crates/paimon/src/table/format_partition.rs

Commands run:

cargo fmt --all -- --check
cargo clippy --locked -p paimon -p paimon-datafusion --all-targets -- -D warnings
cargo test --locked -p paimon --lib --test format_partition_test --test rest_api_test --test rest_catalog_test
cargo test --locked -p paimon-datafusion --lib --test rest_format_partition_sql --test sql_context_tests --test format_table_statistics

API and Format

Additive except for one signature: RESTApi::list_partitions_paged takes a fourth partition_name_pattern: Option<&str>, matching Java RESTApi.listPartitionsPaged. Both in-crate callers pass None. RESTApi::list_partitions_by_name_pattern is new. The Catalog trait is unchanged; the scan reaches the REST client directly, so adding the pattern there can wait for a caller that needs it.

The storage format is unchanged.

Documentation

docs/src/sql.md gains a Format Table Partitions section covering all four statements, the metastore.partitioned-table prerequisite, the ordering guarantees of ADD/DROP, and what repair does and does not touch.

Scope and limitations

• Applies only to partitioned internal REST Format Tables with metastore.partitioned-table=true and a non-engine implementation.
• Several DROP PARTITION specifications are written as repeated clauses, DROP PARTITION (...), DROP PARTITION (...). Hive and Spark write DROP PARTITION (...), (...), which sqlparser 0.62 cannot parse: parse_alter_table comma-separates ALTER operations, and a bare second PARTITION is read as RENAME PARTITION. Accepting the Hive spelling needs either a change in sqlparser or a hand-rolled pre-parse like the one SHOW PARTITIONS already uses. Happy to add the pre-parse if that is preferred.
• Predicate pushdown to listPartitionsByFilter is not included. [spec] Support parsing REST catalog Predicate and Transform JSON #500 added REST predicate JSON parsing but not serialization, so there is no encoder to send yet. The prefix pattern covers the leading-equality case.
• Partition columns typed DECIMAL, TIMESTAMP, FLOAT, DOUBLE or BINARY are still unsupported on this path, as they were before this change. Widening that means aligning the whole string-to-value conversion with TypeUtils.castFromString, which is better done on its own.

[JingsongLi]

1. High: DROP PARTITION is inconsistent with Java semantics

   • Rust only supports single, complete partitions: sql_context.rs:1064 rejects multiple specs, and sql_context.rs:1734 uses require_complete=true.
   • Java supports dropping multiple partitions at once and also supports arbitrary partial specifications such as dt=... and hh=..., which are expanded to include all matching leaf partitions.
   • Impact: Commands executable in Java, such as DROP PARTITION (dt=‘20260715’) and batch DROP operations, fail outright in Rust. Recommendation: Follow Java’s approach—pre-check complete specs using list-by-names; expand partial specs after a single catalog traversal.

2. High: Catalog-managed scans lack partition pruning via the REST endpoint

   • Rust’s format_table_scan.rs:216 unconditionally uses list_partitions to retrieve all partitions in the table before executing the predicate locally.
   • Java FormatTableScan.java:230 extracts the leading equality prefix and the full predicate; CatalogFormatTablePartitionManager.java:68 pushes the pattern/predicate and page size (1000) down to REST.
   • Even when querying a single partition in a table with many partitions, all partition metadata is downloaded and parsed, which may cause significant latency and memory issues.

3. Note: Boolean partition values are incompatible with Java

   • Repair preserves and registers the original values in the catalog, but Rust format_partition.rs:216 uses str::parse::<bool>(), which only accepts lowercase true/false.
   • Java ignores case and accepts t/y/yes/1 and f/n/no/0.
   • Therefore, if a catalog entry active=TRUE is registered via Rust MSCK, subsequent scan or SHOW operations will report “invalid catalog partition metadata”; Java can read it normally. It is recommended to adopt the Java-compatible rules and perform additional cross-platform testing.

[sundapeng]

@JingsongLi Thanks for the review. All three are fixed, and the branch is rebased onto current main (it was conflicting with #600 and #627). Details below, including one place where I could not go all the way and would like your call.

1. DROP PARTITION semantics

DROP PARTITION now takes several specifications per statement, and a specification that fixes only some of the partition keys expands to every registered partition it matches. The keys need not be a leading prefix, so DROP PARTITION (hh = '10') on a (dt, hh) table works, matching PaimonFormatTable.dropFormatTablePartitions where partial specs are matched by arbitrary key subset. One catalog listing serves the whole statement, however many specifications it carries.

Error semantics follow Java as well: a complete specification names one partition, so a missing one raises unless IF EXISTS; a partial one describes a set that is allowed to come out empty, so it is a no-op. A specification that raises leaves the whole statement unapplied.

One thing I could not do. Hive and Spark write several specs as DROP PARTITION (a), (b), and sqlparser 0.62 cannot parse that: parse_alter_table comma-separates ALTER operations, and a bare second PARTITION is read as RENAME PARTITION. So multiple specs are written as repeated clauses:

ALTER TABLE t DROP PARTITION (dt = '20260722'), DROP PARTITION (dt = '20260723');

The alternatives are a change in sqlparser, or a hand-rolled pre-parse of the statement like the one SHOW PARTITIONS already uses here. I did not add the pre-parse because it duplicates the ALTER TABLE grammar for one syntax variant, but I will add it if you want the Hive spelling accepted.

I also did not add list-by-names. Java resolves complete specs through it and asserts in tests that the complete-spec path never does a full traversal, but paimon-rust has no listPartitionsByNames anywhere yet (no Catalog method, no RESTApi method, no resource path, no request type). Adding the whole stack here would grow a PR that is already large. The current code needs the registered spec anyway to resolve the directory, so it costs one listing per statement rather than one per specification. Happy to do it as a follow-up, or here if you prefer.

2. Partition pruning through the REST endpoint

A catalog-managed scan now extracts the leading equality prefix from its filter, builds the partition-name prefix pattern with the same contract as PartitionPathUtils.buildPartitionNamePrefixPattern (escaped key=value joined by /, % the only wildcard, complete prefix means the exact name, shorter prefix gets /%, no pattern when a value is blank or escaping produced a literal %), and sends it as partitionNamePattern with maxResults=1000. The local per-partition match stays, so a catalog that ignores the pattern still produces the same result set.

One Rust-specific detail worth flagging. A straight port of the Java extractor would have pushed nothing in the case that matters most. PartitionFilter::from_predicate collapses a filter that pins every partition key into a PartitionSet and drops the predicate, so WHERE dt = 'a' AND hh = '10' never reaches the scan as a Predicate. The extractor therefore also handles PartitionSet, taking the longest common leading prefix of its rows. A single-partition set gives the exact name, and dt = 'a' AND hh IN ('10', '11') gives dt=a/%.

Predicate pushdown to listPartitionsByFilter is not included: #500 added REST predicate JSON parsing but not serialization, so there is nothing to encode with yet.

3. Boolean partition values

parse_format_partition_value now accepts t/true/y/yes/1 and f/false/n/no/0 case-insensitively, mirroring BinaryStringUtils.toBoolean. There was a unit test asserting that yes is rejected, which is exactly the bug, so it is replaced with a table covering every spelling in both directions plus the rejections.

While confirming this I found the same class of problem is wider than booleans: parse_format_partition_value covers BOOLEAN, the integer types, CHAR/VARCHAR, DATE and TIME, while TypeUtils.castFromString also covers DECIMAL, FLOAT, DOUBLE, TIMESTAMP, TIMESTAMP_LTZ and BINARY. A catalog-managed table partitioned by any of those cannot be scanned. That is not a regression from this PR, and closing it properly means aligning both directions of the conversion, which are currently split across two modules that have already drifted. I would rather do it in its own PR than widen this one. Tell me if you want it here instead.

Also in this push

docs/src/sql.md documents all four statements, and the PR description now follows the template.