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

docs/docs/spark/auxiliary.md

@@ -130,6 +130,11 @@ 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`, while the
+`FOR COLUMNS` forms above are not supported, 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,38 @@ 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. Each measured field replaces
+the one the catalog held, so running it twice reports the same numbers as running it once, while a
+field it could not measure leaves the stored one as it was. 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 a partition holding no files counts as zero, while a format that carries none
+(CSV, TEXT, JSON) leaves the row count the catalog already held rather than guessing one. Reading
+footers costs one open per file, so it runs on the executors and `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

@@ -36,7 +36,7 @@
  <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 measures at once, so that a table with many partitions does not burst listing requests at storage.</td>
+ <td>How many requests MSCK REPAIR TABLE and ANALYZE TABLE use at once to measure Format Table partitions, so that a large table does not burst them at storage.</td>
  </tr>
  <tr>
  <td><h5>legacy-timestamp-mapping.enabled</h5></td>

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

@@ -75,8 +75,10 @@ default void createPartitions(List<Map<String, String>> partitions, boolean igno
  *
  * <p>Statistics are matched to {@code partitions} by {@link PartitionStatistics#spec()} and may
  * cover only some of them; {@code replaceStatistics} says whether they replace what the catalog
- * holds or add to it, and is ignored when {@code statistics} is null. Reporting never
- * unregisters a partition.
+ * holds or add to it, and is ignored when {@code statistics} is null. A field reported as
+ * unknown says nothing about itself and leaves the stored one as it was, so a measurement that
+ * could not take a number does not erase the last one that could. Reporting never unregisters a
+ * partition.
  *
  * <p>This is the method an implementation provides, so that none can report nothing by
  * accident: a decorator that forwards only the two-argument form would otherwise drop every

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

@@ -19,17 +19,22 @@
 package org.apache.paimon.table.format;
 
 import org.apache.paimon.CoreOptions;
-import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.format.FileFormat;
+import org.apache.paimon.format.SimpleStatsExtractor;
 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;
@@ -38,15 +43,16 @@
 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. File
- * count, byte size and last file creation time come from the listing; the row count does not, since
- * no listing opens a file. A partition holding nothing measures as an exact zero on the file
- * numbers, with no last file to date.
+ * Measures what the partitions of a Format Table currently hold. File count, byte size and last
+ * file creation time come from a directory listing; the row count needs every file's footer, which
+ * no listing opens, so it is asked for rather than assumed. A partition holding nothing measures as
+ * an exact zero, with no last file to date.
  *
  * <p>It lists through {@link FormatTableScan#listDataFiles}, the listing the scan itself uses, so a
  * measurement counts exactly the files a reader would return and committer staging trees are pruned
@@ -61,15 +67,35 @@ 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);
  }
 
@@ -81,52 +107,92 @@ public List<PartitionStatistics> collect(List<Map<String, String>> partitions) {
  if (partitions.isEmpty()) {
  return Collections.emptyList();
  }
- int threads = Math.min(parallelism, partitions.size());
+ SimpleStatsExtractor rowCounter = withRecordCount ? rowCounter() : null;
+ if (withRecordCount && rowCounter == null) {
+ LOG.info(
+ "No row counter could be built for format {} of table {}, so the row counts of "
+ + "the measured partitions stay unknown.",
+ table.format(),
+ table.fullName());
+ }
+ // A listing is one request per partition, a footer read one per file, so counting rows
+ // leaves work to spread even when a single partition was asked for.
+ int threads = rowCounter == null ? Math.min(parallelism, partitions.size()) : parallelism;
  if (threads == 1) {
  List<PartitionStatistics> statistics = new ArrayList<>(partitions.size());
  for (Map<String, String> partition : partitions) {
- statistics.add(measure(partition));
+ List<FileStatus> files = listDataFiles(partition);
+ List<Long> rowCounts = new ArrayList<>(files.size());
+ for (FileStatus file : files) {
+ if (rowCounter != null) {
+ rowCounts.add(rowCount(rowCounter, file));
+ }
+ }
+ statistics.add(statistics(partition, files, sum(rowCounts, rowCounter != null)));
  }
  return statistics;
  }
 
  ExecutorService executor =
  ThreadPoolUtils.createCachedThreadPool(threads, "FORMAT-TABLE-STATS-THREAD-POOL");
  try {
- List<Future<PartitionStatistics>> futures = new ArrayList<>(partitions.size());
+ List<Future<List<FileStatus>>> listings = new ArrayList<>(partitions.size());
  for (Map<String, String> partition : partitions) {
- futures.add(executor.submit(() -> measure(partition)));
+ listings.add(executor.submit(() -> listDataFiles(partition)));
+ }
+ List<List<FileStatus>> files = new ArrayList<>(partitions.size());
+ for (Future<List<FileStatus>> listing : listings) {
+ files.add(await(listing));
+ }
+ // Every file of every partition goes to the same pool, so one partition holding many
+ // files is counted with all of it rather than with one thread of it.
+ List<List<Future<Long>>> rowCounts = new ArrayList<>(partitions.size());
+ for (List<FileStatus> partitionFiles : files) {
+ List<Future<Long>> counts = new ArrayList<>(partitionFiles.size());
+ for (FileStatus file : partitionFiles) {
+ if (rowCounter != null) {
+ counts.add(executor.submit(() -> rowCount(rowCounter, file)));
+ }
+ }
+ rowCounts.add(counts);
  }
  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());
+ for (int i = 0; i < partitions.size(); i++) {
+ List<Long> counted = new ArrayList<>(rowCounts.get(i).size());
+ for (Future<Long> count : rowCounts.get(i)) {
+ counted.add(await(count));
  }
+ statistics.add(
+ statistics(
+ partitions.get(i), files.get(i), sum(counted, rowCounter != null)));
  }
  return statistics;
  } finally {
  executor.shutdownNow();
  }
  }
 
- private PartitionStatistics measure(Map<String, String> partition) {
- FileIO fileIO = table.fileIO();
+ private <T> T await(Future<T> future) {
+ try {
+ return 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());
+ }
+ }
+
+ /** The data files of a partition; a registered partition whose directory is gone has none. */
+ private List<FileStatus> listDataFiles(Map<String, String> partition) {
  Path partitionPath = partitionPath(partition);
- List<FileStatus> files;
  try {
  // A missing directory surfaces here as a FileNotFoundException, so it needs no
  // separate existence check.
-files = FormatTableScan.listDataFiles(fileIO, partitionPath);
+return FormatTableScan.listDataFiles(table.fileIO(), partitionPath);
  } catch (FileNotFoundException e) {
- // A registered partition whose directory is gone reads as empty.
- return empty(partition);
+ return Collections.emptyList();
  } catch (IOException e) {
  throw new UncheckedIOException(
  String.format(
@@ -136,37 +202,90 @@ private PartitionStatistics measure(Map<String, String> partition) {
  partitionPath, table.fullName()),
  e);
  }
+ }
+
+ /**
+ * The rows the counted files hold. One file whose footer could not be read makes the whole
+ * partition unknown rather than short: a sum missing a file, reported as exact, is worse than
+ * no number at all. A measurement that counted nothing knows no row count, while one that
+ * counted an empty partition knows it holds none.
+ */
+ private static long sum(List<Long> rowCounts, boolean counted) {
+ if (!counted) {
+ return PartitionStatistics.UNKNOWN;
+ }
+ long rows = 0;
+ for (long rowCount : rowCounts) {
+ if (!PartitionStatistics.isKnown(rowCount)) {
+ return PartitionStatistics.UNKNOWN;
+ }
+ rows += rowCount;
+ }
+ return rows;
+ }
 
- long fileCount = 0;
+ /** What the listed files of a partition add up to. */
+ private static PartitionStatistics statistics(
+ Map<String, String> partition, List<FileStatus> files, long recordCount) {
  long fileSizeInBytes = 0;
- long lastFileCreationTime = 0;
+ long lastFileCreationTime = PartitionStatistics.UNKNOWN;
  for (FileStatus file : files) {
- fileCount++;
  fileSizeInBytes += file.getLen();
  lastFileCreationTime = Math.max(lastFileCreationTime, file.getModificationTime());
  }
- if (fileCount == 0) {
- return empty(partition);
- }
  return new PartitionStatistics(
  partition,
- // A listing never opens a file, so the rows a partition holds stay unknown.
- PartitionStatistics.UNKNOWN,
+ recordCount,
  fileSizeInBytes,
-fileCount,
+files.size(),
  lastFileCreationTime,
  PartitionStatistics.UNKNOWN_TOTAL_BUCKETS);
  }
 
- /** A partition with nothing in it: the file numbers are an exact zero. */
- private static PartitionStatistics empty(Map<String, String> partition) {
- return new PartitionStatistics(
- partition,
- PartitionStatistics.UNKNOWN,
- 0L,
- 0L,
- PartitionStatistics.UNKNOWN,
- 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;
+ }
  }
 
  private Path partitionPath(Map<String, String> partition) {