[core][spark] Support ANALYZE TABLE on catalog-managed format tables - #9298
[core][spark] Support ANALYZE TABLE on catalog-managed format tables#9298sundapeng wants to merge 1 commit into
Conversation
sundapeng
changed the title
sundapeng
changed the title
140cff9 to
573e681
Compare
573e681 to
f24cdad
Compare
b8719f9 to
80eb40b
Compare
The catalog statistics of a Format Table are written by whoever touched it: a commit reports what it wrote, and MSCK REPAIR TABLE measures what it registers. Neither answers for a partition written by something the catalog never saw, and Spark rejects ANALYZE TABLE for every v2 table in the analyzer, so no statement recomputed one. ANALYZE TABLE t [PARTITION (...)] COMPUTE STATISTICS [NOSCAN] now measures the registered partitions and writes the result back. NOSCAN stops at the directory listing - file count, byte size, last file creation time - while a full ANALYZE also reads each file footer for its row count, exact for the formats that carry one and left as it was for the ones that do not. A PARTITION (...) clause selects the partitions whose leading values it fixes, the shape the catalog can select on, and naming a partition the table does not have is a NoSuchPartitionException as it is on any other table. Analyzing measures partitions and never adds or removes one. A Format Table has nowhere to keep column statistics, so FOR [ALL] COLUMNS keeps Spark's own rejection, and so does a table discovering its partitions from the filesystem, which has no catalog to write to. Listing a partition is one request and reading a footer is one per file, and both go to the same pool, so format-table.statistics.parallelism applies to a single large partition as much as to many small ones.
80eb40b to
ea9333b
Compare
sundapeng
changed the title
| valueByKey.filter(_._2.isDefined).map(_._1).mkString("[", ", ", "]")) | ||
| } | ||
| // Kept in partition-key order, so a message built from it reads in that order too. | ||
| ListMap(prefix.map { case (key, value) => key -> value.get }: _*) |
There was a problem hiding this comment.
[P2] normalizePartitionSpec only resolves partition column names; it does not cast or canonicalize their values, so this prefix still contains the raw parser strings. For example, an INT partition registered as p=1 is not found by ANALYZE ... PARTITION (p = '01'), because the catalog is queried with p=01 and the command throws NoSuchPartitionException. Null or empty values likewise are not converted to the configured default partition name. Please cast the specified values using v2Table.partitionSchema under Spark SQL semantics and reuse the existing toPaimonPartition/InternalRowPartitionComputer conversion before building the catalog prefix; tests should cover numeric canonicalization and null/default partitions.
| List<Future<Long>> counts = new ArrayList<>(partitionFiles.size()); | ||
| for (FileStatus file : partitionFiles) { | ||
| if (rowCounter != null) { | ||
| counts.add(executor.submit(() -> rowCount(rowCounter, file))); |
There was a problem hiding this comment.
[P2] Please run full-ANALYZE footer reads on Spark executors. This parallelizes files, but every footer open still runs in a driver-local pool, and the driver eagerly retains one FileStatus plus one Future for every data file before aggregation. A format table with hundreds of thousands or millions of files can therefore make ANALYZE driver-bound or exhaust the driver heap while executor capacity is idle. The Spark layer could batch file descriptors and process them with RDD mapPartitions, creating FileIO and the stats extractor once per task and reducing partial results per catalog partition; format-table.statistics.parallelism can bound the RDD partitions/storage concurrency. The engine-neutral footer parsing and merge logic can remain in paimon-core, and NOSCAN can keep the local path.
2 participants
Purpose
The catalog statistics of a Format Table are written by whoever touched it: a commit reports what
it wrote, and
MSCK REPAIR TABLEmeasures what it registers. Neither answers for a partitionwritten by something the catalog never saw, and there was no statement that recomputes one, because
Spark rejects
ANALYZE TABLEfor every v2 table in the analyzer.ANALYZE TABLE t [PARTITION (...)] COMPUTE STATISTICS [NOSCAN]now measures the registeredpartitions of a Format Table and writes the result back.
NOSCANstops at the directory listing -file count, byte size, last file creation time - while a full ANALYZE also reads each file footer
for its row count, exact for the formats that carry one and left as it was for the ones that do
not. A
PARTITION (...)clause selects the partitions whose leading values it fixes, the shape thecatalog can select on; naming a partition the table does not have is
NoSuchPartitionException,as on any other table.
Analyzing measures partitions and never adds or removes one -
MSCK REPAIR TABLEis what doesthat. A Format Table has nowhere to keep column statistics, so
FOR [ALL] COLUMNSkeeps Spark'sown rejection, and so does a table discovering its partitions from the filesystem, which has no
catalog to write to.
Listing a partition is one request, reading a footer is one per file, and both go to the same pool,
so
format-table.statistics.parallelismapplies to a single large partition as much as to manysmall ones.
Tests
FormatTablePartitionStatsCollectorTestcovers the measurement: exact row counts, an unreadablefooter poisoning only the partition it is in, an empty partition counting as zero, and counts
staying with their partition when the pool measures several at once.
CatalogManagedPartitionAnalyzeTestcovers the statement end to end against the semantics Sparkgives it on a metastore table - partition selection, name resolution under both case sensitivities,
NOSCAN, repeated runs, and the forms that are rejected.