feat(datafusion): support ANALYZE TABLE on catalog-managed format tables - #815

Purpose

Child of #591, which stays open as a draft that shows the full feature. A Format Table loaded from a REST catalog with metastore.partitioned-table=true has its partitions registered in the catalog, and the catalog can hold statistics for each of them: record count, file count, size and last file creation time. The Rust client had no way to measure a partition and report them. This adds ANALYZE TABLE for such tables, following the Java PaimonAnalyzeFormatTablePartitionsCommand and FormatTablePartitionStatsCollector.

Depends on #817 , which depends on #816. Until they merge, this branch includes their five commits; please review only the last two: feat(datafusion): support ANALYZE TABLE on catalog-managed format tables and docs(datafusion): document ANALYZE TABLE for format tables.

This PR was rebased onto the restructured stack. ANALYZE itself is unchanged: its tests now use the shared helpers of #816, and a value in PARTITION (...) is read by the shared parse_format_partition_value that #816 introduces for partition literals.

Brief change log

• ANALYZE TABLE t [PARTITION (...)] COMPUTE STATISTICS [NOSCAN] on a Format Table with catalog-managed partitions measures the registered partitions from storage and reports them through Catalog::create_partitions_with_statistics with replaceStatistics=true, so each measured field replaces what the catalog holds. It never adds or removes a partition.
  • NOSCAN reports the file count, total size and latest file modification time, and leaves the row count unknown (-1).
  • A full ANALYZE also reads Parquet and ORC footers for row counts. A footer that cannot be read leaves the row count of its partition unknown rather than short, a partition without files holds exactly zero rows, and other formats report an unknown row count.
  • PARTITION (...) selects a leading run of partition values; a column named without a value means every value of it. A prefix with no registered partition is an error.
  • Refused: FOR COLUMNS, CACHE METADATA, a missing COMPUTE STATISTICS, tables whose partitions the catalog does not manage, an empty or whitespace-only string in PARTITION (...) for a string column (it would name the default partition, as ADD and DROP in feat(datafusion): add SHOW, ADD and DROP PARTITION for catalog-managed format tables #816 refuse), and any selected partition at a custom location.
  • format-table.statistics.parallelism (a session setting, SET 'paimon.<key>', as in Java; default 8, at least 1) bounds the listings and footer reads in flight.
• FormatTablePartitionStatsCollector, public in paimon::table, does the measuring. A listing failure fails the whole collection, since a truncated listing cannot be told apart from a partition that lost files.
• The Format Table scan's file listing is factored into list_format_table_data_files, which the collector shares, so a measurement counts exactly the files a scan of that partition reads and leaves committer staging trees out (the rule from fix(table): skip staging files and list format table partitions concurrently #813 ). The scan's behaviour is unchanged.
• read_file_row_count reads the row count of a Parquet or ORC file from its footer without decoding rows.

Follow-up: using the reported row counts in DataFusion scan statistics, as apache/paimon#9351 does for Spark.

Tests

• crates/integrations/datafusion/tests/rest_format_partition_sql.rs:
  • NOSCAN and full measurement replacing what the catalog holds, and a later NOSCAN keeping the known row counts;
  • PARTITION selecting a leading run of values, and its errors;
  • a partition value read as its column type;
  • staging trees, markers, hidden files and non-data files left out by both ANALYZE and a scan;
  • an unreadable footer leaving the row count unknown;
  • refusals: FOR COLUMNS, a blank string partition value, a partition at a custom location, a non-positive parallelism read as one, and a table without catalog-managed partitions.
• format_table_scan::tests::test_data_file_listing_returns_what_a_partition_scan_reads for the shared listing.
• The compile-time check that the future of SQLContext::sql stays Send now covers the new statement too.

Commands run on the top commit:

cargo fmt --all -- --check
cargo clippy --locked --all-targets --workspace --features fulltext,vortex -- -D warnings
cargo test --locked -p paimon --all-targets --features fulltext,vortex
cargo test --locked --no-fail-fast -p paimon-datafusion --all-targets
cargo test --locked -p paimon-rest-server --all-targets

fmt and clippy pass. paimon: 2967 tests pass. paimon-rest-server: 10 tests pass. paimon-datafusion: 780 pass, including the 15 tests in rest_format_partition_sql.rs, and 39 fail. The failures are the 39 tests that also fail by name on main in this environment, because they read fixture tables provisioned by make docker-up (two also need the lumina native library).

Additions without a direct Java counterpart

• read_file_row_count reads a Parquet or ORC file's row count from its footer. Java counts rows with the format's SimpleStatsExtractor, which the Rust client does not have for Format Tables.
• A full ANALYZE measures in process with bounded concurrency, where Spark spreads the footer reads over executors (PaimonAnalyzeFormatTablePartitionsCommand.measureOnExecutors).

API and Format

• New public paimon::table::FormatTablePartitionStatsCollector.
• New option format-table.statistics.parallelism, read by SQLContext.

The storage format is unchanged.

Documentation

docs/src/sql.md gains an ANALYZE TABLE subsection under Format Table Partitions and lists ANALYZE TABLE in the SQL support scope.