[WIP][spark] Support ANALYZE TABLE on catalog-managed format tables

docs/docs/spark/auxiliary.md

@@ -130,6 +130,10 @@ ANALYZE TABLE my_table COMPUTE STATISTICS FOR COLUMNS col1;
 ANALYZE TABLE my_table COMPUTE STATISTICS FOR ALL COLUMNS;
 ```
 
+On a Format Table with catalog-managed partitions the statement means something narrower: it
+measures the table's partitions and supports `PARTITION (...)` and `NOSCAN`, see
+[Manage Format Table Partitions](./sql-ddl#manage-format-table-partitions).
+
 ## Refresh table
 
 The REFRESH TABLE statement invalidates the cached entries, which include data and metadata of the given table.

docs/docs/spark/sql-ddl.md

@@ -216,15 +216,35 @@ ALTER TABLE my_table ADD PARTITION (dt='2025-01-01');
 ALTER TABLE my_table DROP PARTITION (dt='2025-01-01');
 MSCK REPAIR TABLE my_table;
 SHOW PARTITIONS my_table;
+ANALYZE TABLE my_table PARTITION (dt='2025-01-01') COMPUTE STATISTICS NOSCAN;
 ```
 
 On a Format Table whose partitions are discovered from the filesystem, `ADD PARTITION`,
-`DROP PARTITION` and `MSCK REPAIR TABLE` fail with an error.
+`DROP PARTITION`, `MSCK REPAIR TABLE` and `ANALYZE TABLE` fail with an error.
 
 `ADD PARTITION` creates the partition directory and registers the partition; querying a newly
 added partition before any data is written returns no rows. `DROP PARTITION` unregisters the
 partition and deletes its directory.
 
+`ANALYZE TABLE` measures partitions. A Format Table has no snapshot to carry a table-level
+statistic and no column statistics, so `COMPUTE STATISTICS FOR COLUMNS` and `FOR ALL COLUMNS` are
+not supported on it; what the statement writes back to the catalog is the file count, byte size,
+last file creation time and row count of the partitions it measured. The measurement replaces
+what the catalog held for those partitions, so running it twice reports the same numbers as
+running it once, and it never adds or removes a partition — use `MSCK REPAIR TABLE` for that.
+
+`NOSCAN` stops at the directory listing, which gives everything except the row count. Without it,
+the row count is read from each file's footer, so it is exact for the formats that carry one
+(Parquet, ORC) and stays unknown for the ones that do not (CSV, TEXT, JSON) rather than being
+guessed. Reading footers costs one open per file, so `NOSCAN` is the cheaper of the two.
+
+A `PARTITION (...)` clause must give values for a leading run of the partition columns, because
+that is the shape the catalog can select on. On a table partitioned by `(dt, hh)`,
+`PARTITION (dt='2025-01-01')` and `PARTITION (dt='2025-01-01', hh)` both measure every hour of
+that day, while `PARTITION (hh='01')` is rejected rather than widened to every day. Naming a
+partition that is not registered is an error too, rather than a statement that reports success for
+having measured nothing.
+
 :::info
 
 `metastore.partitioned-table = true` enables catalog-managed partitions, which requires an

docs/generated/spark_connector_configuration.html

@@ -26,6 +26,18 @@
         </tr>
     </thead>
     <tbody>
+        <tr>
+            <td><h5>format-table.repair.collect-statistics</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>Whether MSCK REPAIR TABLE on a Format Table also measures the partitions it finds and reports their statistics to the catalog. Off by default, because it changes what a repair costs: the plain repair lists partition directories, while measuring lists the files inside every one of them.</td>
+        </tr>
+        <tr>
+            <td><h5>format-table.statistics.parallelism</h5></td>
+            <td style="word-wrap: break-word;">8</td>
+            <td>Integer</td>
+            <td>How many Format Table partitions MSCK REPAIR TABLE and ANALYZE TABLE measure at once. Listing a partition is one round trip to storage, so the driver spends its time waiting; the ceiling keeps that from turning into a burst of requests against a table with many partitions. This is the driver-side measuring path, separate from 'format-table.scan.list-parallelism', which governs the same per-partition listing on the read path.</td>
+        </tr>
         <tr>
             <td><h5>legacy-timestamp-mapping.enabled</h5></td>
             <td style="word-wrap: break-word;">false</td>

paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionStatsCollector.java

@@ -0,0 +1,305 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table.format;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.format.FileFormat;
+import org.apache.paimon.format.SimpleStatsExtractor;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.FileStatus;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.partition.PartitionStatistics;
+import org.apache.paimon.statistics.SimpleColStatsCollector;
+import org.apache.paimon.table.FormatTable;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.PartitionPathUtils;
+import org.apache.paimon.utils.ThreadPoolUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+
+/**
+ * Measures what the partitions of a Format Table currently hold, by listing their directories.
+ *
+ * <p>File count, byte size and last file creation time come from the listing itself. A listing only
+ * reports a modification time, which for the immutable data files of a Format Table is when the
+ * file was created; a file rewritten in place therefore dates its partition by the rewrite. This is
+ * a different clock from the one a managed table records in {@link
+ * org.apache.paimon.manifest.PartitionEntry}, which carries the creation time the writer stamped.
+ * The row count needs the file footer and is only available for the formats that carry one — a CSV,
+ * TEXT or JSON partition keeps an unknown row count rather than a guessed one. A partition that
+ * holds nothing, whether its directory is empty or gone, measures as an exact zero on the three
+ * counts; its last file creation time stays unknown, because there is no last file to date.
+ *
+ * <p>It measures through {@link FormatTableScan#listDataFiles}, the listing the scan itself uses,
+ * so a measurement counts exactly the files a reader would return. Committer staging trees are
+ * pruned rather than walked: the files under them carry ordinary data file names, and a measurement
+ * that counted them would report a partition holding more than any query returns.
+ *
+ * <p>The result is a whole-partition measurement, so it is reported as a replacement rather than an
+ * increment. This never decides that a partition should exist or stop existing: it only measures
+ * the ones it is given.
+ *
+ * <p>Listing failures abort the whole collection rather than yield a partial measurement — a
+ * truncated listing looks exactly like a partition that lost files.
+ */
+public class FormatTablePartitionStatsCollector {
+
+    private static final Logger LOG =
+            LoggerFactory.getLogger(FormatTablePartitionStatsCollector.class);
+
+    /** Row counts that need no columns: the file footer alone answers how many rows it holds. */
+    private static final RowType NO_COLUMNS = RowType.builder().build();
+
+    private static final SimpleColStatsCollector.Factory[] NO_COLLECTORS =
+            new SimpleColStatsCollector.Factory[0];
+
+    private final FormatTable table;
+
+    private final boolean onlyValueInPath;
+    private final int parallelism;
+
+    private final boolean withRecordCount;
+
+    /** Measures from the listing alone, leaving the record count unknown. */
+    public FormatTablePartitionStatsCollector(FormatTable table, int parallelism) {
+        this(table, false, parallelism);
+    }
+
+    /**
+     * Measures from the listing, and when {@code withRecordCount} is set also opens every file's
+     * footer for the rows it holds. That is the expensive half, so it is asked for rather than
+     * assumed.
+     */
+    public FormatTablePartitionStatsCollector(
+            FormatTable table, boolean withRecordCount, int parallelism) {
+        this.table = table;
+        this.onlyValueInPath =
+                new CoreOptions(table.options()).formatTablePartitionOnlyValueInPath();
+        this.withRecordCount = withRecordCount;
+        this.parallelism = Math.max(1, parallelism);
+    }
+
+    /**
+     * Measures the given partitions. The result is aligned to {@code partitions} one for one, so a
+     * caller can send it straight to the catalog alongside the same specs.
+     */
+    public List<PartitionStatistics> collect(List<Map<String, String>> partitions) {
+        if (partitions.isEmpty()) {
+            return Collections.emptyList();
+        }
+        SimpleStatsExtractor rowCounter = withRecordCount ? rowCounter() : null;
+        if (withRecordCount && rowCounter == null) {
+            LOG.info(
+                    "Format {} of table {} carries no row count in its files, so the row counts of "
+                            + "the measured partitions stay unknown.",
+                    table.format(),
+                    table.fullName());
+        }
+        int threads = Math.min(parallelism, partitions.size());
+        if (threads == 1) {
+            List<PartitionStatistics> statistics = new ArrayList<>(partitions.size());
+            for (Map<String, String> partition : partitions) {
+                statistics.add(measure(partition, rowCounter));
+            }
+            return statistics;
+        }
+
+        ExecutorService executor =
+                ThreadPoolUtils.createCachedThreadPool(threads, "FORMAT-TABLE-STATS-THREAD-POOL");
+        try {
+            List<Future<PartitionStatistics>> futures = new ArrayList<>(partitions.size());
+            for (Map<String, String> partition : partitions) {
+                futures.add(executor.submit(() -> measure(partition, rowCounter)));
+            }
+            List<PartitionStatistics> statistics = new ArrayList<>(partitions.size());
+            for (Future<PartitionStatistics> future : futures) {
+                try {
+                    statistics.add(future.get());
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    throw new RuntimeException(
+                            "Interrupted while measuring partitions of table " + table.fullName(),
+                            e);
+                } catch (ExecutionException e) {
+                    throw asRuntime(e.getCause());
+                }
+            }
+            return statistics;
+        } finally {
+            executor.shutdownNow();
+        }
+    }
+
+    private PartitionStatistics measure(
+            Map<String, String> partition, @Nullable SimpleStatsExtractor rowCounter) {
+        FileIO fileIO = table.fileIO();
+        Path partitionPath = partitionPath(partition);
+        List<FileStatus> files;
+        try {
+            // The same listing the scan uses, so a measurement counts exactly what a reader would
+            // return: staging trees are pruned rather than walked, and no separate existence check
+            // is needed because a missing directory surfaces here as a FileNotFoundException.
+            files = FormatTableScan.listDataFiles(fileIO, partitionPath);
+        } catch (FileNotFoundException e) {
+            // A registered partition whose directory is gone reads as empty, so measuring it as an
+            // exact zero says the same thing the reader already does.
+            return empty(partition, rowCounter);
+        } catch (IOException e) {
+            throw new UncheckedIOException(
+                    String.format(
+                            "Failed to list partition %s of table %s; no statistics are written "
+                                    + "because a partial listing cannot be told apart from a "
+                                    + "partition that lost files.",
+                            partitionPath, table.fullName()),
+                    e);
+        }
+
+        long fileCount = 0;
+        long fileSizeInBytes = 0;
+        long lastFileCreationTime = 0;
+        long recordCount = rowCounter == null ? PartitionStatistics.UNKNOWN : 0L;
+        for (FileStatus file : files) {
+            fileCount++;
+            fileSizeInBytes += file.getLen();
+            lastFileCreationTime = Math.max(lastFileCreationTime, file.getModificationTime());
+            if (PartitionStatistics.isKnown(recordCount)) {
+                long rows = rowCount(rowCounter, file);
+                recordCount =
+                        PartitionStatistics.isKnown(rows)
+                                ? recordCount + rows
+                                : PartitionStatistics.UNKNOWN;
+            }
+        }
+        if (fileCount == 0) {
+            return empty(partition, rowCounter);
+        }
+        return new PartitionStatistics(
+                partition,
+                recordCount,
+                fileSizeInBytes,
+                fileCount,
+                lastFileCreationTime,
+                PartitionStatistics.UNKNOWN_TOTAL_BUCKETS);
+    }
+
+    /**
+     * Rows in one file, or unknown when its footer cannot be read. One unreadable file makes the
+     * whole partition unknown rather than short: a sum missing a file, reported as exact, is worse
+     * than no number at all.
+     */
+    private long rowCount(SimpleStatsExtractor rowCounter, FileStatus file) {
+        try {
+            return rowCounter
+                    .extractWithFileInfo(table.fileIO(), file.getPath(), file.getLen())
+                    .getRight()
+                    .getRowCount();
+        } catch (Exception e) {
+            LOG.warn(
+                    "Failed to read the row count of {} in table {}; the row count of its "
+                            + "partition stays unknown.",
+                    file.getPath(),
+                    table.fullName(),
+                    e);
+            return PartitionStatistics.UNKNOWN;
+        }
+    }
+
+    /**
+     * A footer reader for this table's format, or null when the format carries no row count. It is
+     * built with no columns on purpose: only the file's row count is wanted, and asking for column
+     * statistics would both cost more and make the reader depend on the file schema matching the
+     * table's.
+     */
+    @Nullable
+    private SimpleStatsExtractor rowCounter() {
+        try {
+            CoreOptions options = new CoreOptions(table.options());
+            Optional<SimpleStatsExtractor> extractor =
+                    FileFormat.fileFormat(options).createStatsExtractor(NO_COLUMNS, NO_COLLECTORS);
+            return extractor.orElse(null);
+        } catch (Exception e) {
+            LOG.warn(
+                    "Failed to create a row counter for format {} of table {}; row counts stay "
+                            + "unknown.",
+                    table.format(),
+                    table.fullName(),
+                    e);
+            return null;
+        }
+    }
+
+    /**
+     * A partition holding nothing. The counts are an exact zero, but there is no last file to date,
+     * so that one field stays unknown rather than claiming the epoch.
+     */
+    /** A partition with nothing in it: the file numbers are an exact zero. */
+    private static PartitionStatistics empty(
+            Map<String, String> partition, @Nullable SimpleStatsExtractor rowCounter) {
+        return new PartitionStatistics(
+                partition,
+                // Only a measurement that counts rows has learned that this partition holds none;
+                // a zero from one that never opens a file would outlive the emptiness, since the
+                // measurement after the files arrive reports unknown and unknown replaces nothing.
+                rowCounter == null ? PartitionStatistics.UNKNOWN : 0L,
+                0L,
+                0L,
+                PartitionStatistics.UNKNOWN,
+                PartitionStatistics.UNKNOWN_TOTAL_BUCKETS);
+    }
+
+    private Path partitionPath(Map<String, String> partition) {
+        LinkedHashMap<String, String> ordered = new LinkedHashMap<>();
+        for (String key : table.partitionKeys()) {
+            if (!partition.containsKey(key)) {
+                throw new IllegalArgumentException(
+                        String.format(
+                                "Partition %s of table %s does not give a value for partition key "
+                                        + "%s, so its directory cannot be located.",
+                                partition, table.fullName(), key));
+            }
+            ordered.put(key, partition.get(key));
+        }
+        return new Path(
+                table.location(),
+                PartitionPathUtils.generatePartitionPathUtil(ordered, onlyValueInPath));
+    }
+
+    private static RuntimeException asRuntime(Throwable cause) {
+        if (cause instanceof RuntimeException) {
+            return (RuntimeException) cause;
+        }
+        return new RuntimeException(cause);
+    }
+}