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

[sundapeng]

Purpose

Last child PR split out of #591. 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.

#816 and #817 have merged, so this PR now applies to main on its own.

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 introduced 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. Only a partition directory that the listing itself reports as not found counts as empty, as in Java; an existence check is not trusted, because on object stores a HEAD on the directory key returns 404 even when files lie below it.
• 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). A scan reads the same files as before, except that a listing failure is no longer read as an empty directory when an existence check on the root returns false.
• read_file_row_count reads the row count of a Parquet or ORC file from its footer without decoding rows.
• The statement lives in crates/integrations/datafusion/src/format_partition_analyze.rs and reuses the partition-spec helpers of format_partition_ddl.rs from feat(datafusion): add SHOW, ADD and DROP PARTITION for catalog-managed format tables #816; sql_context.rs only dispatches it.

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, and test_listing_failure_is_not_read_as_an_empty_directory for a listing that fails before or after returning files.
• 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: 3000 tests pass. paimon-rest-server: 10 tests pass. paimon-datafusion: 781 pass, including the 16 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 from the SQLContext session.

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.

[JingsongLi]

[P2] Propagate listing failures before replacing partition statistics

The reused helper returns an empty list after any listing error when file_io.exists(root) is false. In the locked OpenDAL 0.58.2, OSS/S3 exists performs HEAD on the exact key: table/dt=a can return 404 even when table/dt=a/part.parquet exists. If LIST or a subsequent page fails and HEAD returns 404, full ANALYZE treats the partition as empty and replaces its record count, file count, and byte size with zero; NOSCAN also replaces the latter two with zero.

I reproduced this with fault injection: listing returned one file and then an error, while stat on the partition prefix returned NotFound. collect() returned Ok with all three statistics set to zero.

Please treat only an actual NotFound from the listing as an empty partition and propagate other listing failures before writing statistics back to the catalog.

[sundapeng]

Good catch, and thanks for the fault-injection repro. Only a not-found root now reads as an empty partition; any other listing failure is returned before statistics are written, which also covers the scan path (1f87213).

[JingsongLi]

+1