From f3284f34ff577825aa8f977474882dc1a7d22756 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dapeng=20Sun=28=E5=AD=99=E5=A4=A7=E9=B9=8F=29?= Date: Sun, 9 Aug 2026 02:01:32 +0800 Subject: [PATCH 1/2] [core][spark] Support ANALYZE TABLE on catalog-managed format tables 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. --- docs/docs/spark/auxiliary.md | 5 + docs/docs/spark/sql-ddl.md | 24 +- .../spark_connector_configuration.html | 2 +- .../format/FormatTablePartitionManager.java | 6 +- .../FormatTablePartitionStatsCollector.java | 205 +++++-- ...ormatTablePartitionStatsCollectorTest.java | 154 +++++- .../paimon/spark/SparkConnectorOptions.java | 5 +- .../catalyst/analysis/PaimonAnalysis.scala | 12 +- ...nAnalyzeFormatTablePartitionsCommand.scala | 131 +++++ .../spark/format/PaimonFormatTable.scala | 9 + .../CatalogManagedPartitionAnalyzeTest.scala | 504 ++++++++++++++++++ 11 files changed, 1006 insertions(+), 51 deletions(-) create mode 100644 paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeFormatTablePartitionsCommand.scala create mode 100644 paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionAnalyzeTest.scala diff --git a/docs/docs/spark/auxiliary.md b/docs/docs/spark/auxiliary.md index c046bf12563c..3141728b18fd 100644 --- a/docs/docs/spark/auxiliary.md +++ b/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. diff --git a/docs/docs/spark/sql-ddl.md b/docs/docs/spark/sql-ddl.md index 4ac28fd6521e..336705206a58 100644 --- a/docs/docs/spark/sql-ddl.md +++ b/docs/docs/spark/sql-ddl.md @@ -216,15 +216,37 @@ 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 `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 diff --git a/docs/generated/spark_connector_configuration.html b/docs/generated/spark_connector_configuration.html index 71bb3861488a..d80d14f258ec 100644 --- a/docs/generated/spark_connector_configuration.html +++ b/docs/generated/spark_connector_configuration.html @@ -36,7 +36,7 @@
format-table.statistics.parallelism
8 Integer - 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. + 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.
legacy-timestamp-mapping.enabled
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionManager.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionManager.java index 0e2c5c938533..aaacebbe7159 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionManager.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionManager.java @@ -75,8 +75,10 @@ default void createPartitions(List> partitions, boolean igno * *

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. * *

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 diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionStatsCollector.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionStatsCollector.java index 0f2323546b94..09f1eceea588 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionStatsCollector.java +++ b/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. * *

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,11 +107,28 @@ public List collect(List> 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 statistics = new ArrayList<>(partitions.size()); for (Map partition : partitions) { - statistics.add(measure(partition)); + List files = listDataFiles(partition); + List 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; } @@ -93,22 +136,35 @@ public List collect(List> partitions) { ExecutorService executor = ThreadPoolUtils.createCachedThreadPool(threads, "FORMAT-TABLE-STATS-THREAD-POOL"); try { - List> futures = new ArrayList<>(partitions.size()); + List>> listings = new ArrayList<>(partitions.size()); for (Map partition : partitions) { - futures.add(executor.submit(() -> measure(partition))); + listings.add(executor.submit(() -> listDataFiles(partition))); + } + List> files = new ArrayList<>(partitions.size()); + for (Future> 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>> rowCounts = new ArrayList<>(partitions.size()); + for (List partitionFiles : files) { + List> counts = new ArrayList<>(partitionFiles.size()); + for (FileStatus file : partitionFiles) { + if (rowCounter != null) { + counts.add(executor.submit(() -> rowCount(rowCounter, file))); + } + } + rowCounts.add(counts); } List statistics = new ArrayList<>(partitions.size()); - for (Future 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 counted = new ArrayList<>(rowCounts.get(i).size()); + for (Future count : rowCounts.get(i)) { + counted.add(await(count)); } + statistics.add( + statistics( + partitions.get(i), files.get(i), sum(counted, rowCounter != null))); } return statistics; } finally { @@ -116,17 +172,27 @@ public List collect(List> partitions) { } } - private PartitionStatistics measure(Map partition) { - FileIO fileIO = table.fileIO(); + private T await(Future 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 listDataFiles(Map partition) { Path partitionPath = partitionPath(partition); - List 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 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 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 partition, List 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 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 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 partition) { diff --git a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTablePartitionStatsCollectorTest.java b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTablePartitionStatsCollectorTest.java index 522288c1e018..9ac5953b5fb8 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTablePartitionStatsCollectorTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTablePartitionStatsCollectorTest.java @@ -20,6 +20,10 @@ import org.apache.paimon.CoreOptions; import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.format.FileFormat; +import org.apache.paimon.format.FormatWriter; +import org.apache.paimon.format.FormatWriterFactory; import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.FileStatus; import org.apache.paimon.fs.Path; @@ -286,8 +290,133 @@ void testASpecMissingAPartitionKeyIsRejected() { .hasMessageContaining("month"); } + @Test + void testAStagedPlaceholderDoesNotEraseTheRowCount() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + writeParquet(fileIO, tablePath, PARTITION_DIR + "/data-0.parquet", 3); + // The magic committer's zero-byte placeholder, under a data file name. Reading it as a + // data file fails, and a failed read turns the whole partition's row count into an + // unknown, so pruning the staging tree is what keeps the count exact. + write( + fileIO, + tablePath, + PARTITION_DIR + "/__magic_job-1/tasks/attempt_1/__base/part-1.parquet", + 0); + + PartitionStatistics measured = + new FormatTablePartitionStatsCollector(parquetTable(fileIO, tablePath), true, 1) + .collect(Collections.singletonList(spec("2025", "10"))) + .get(0); + + assertThat(measured.recordCount()).isEqualTo(3); + assertThat(measured.fileCount()).isEqualTo(1); + } + + @Test + void testParquetRowCountsAreMeasuredExactly() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + writeParquet(fileIO, tablePath, PARTITION_DIR + "/data-0.parquet", 3); + writeParquet(fileIO, tablePath, PARTITION_DIR + "/data-1.parquet", 5); + + PartitionStatistics measured = + new FormatTablePartitionStatsCollector(parquetTable(fileIO, tablePath), true, 1) + .collect(Collections.singletonList(spec("2025", "10"))) + .get(0); + + assertThat(measured.recordCount()).isEqualTo(8); + assertThat(measured.fileCount()).isEqualTo(2); + assertThat(measured.fileSizeInBytes()).isPositive(); + assertThat(PartitionStatistics.isKnown(measured.lastFileCreationTime())).isTrue(); + } + + @Test + void testAnUnreadableFooterLeavesTheWholePartitionRowCountUnknown() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + writeParquet(fileIO, tablePath, PARTITION_DIR + "/data-0.parquet", 3); + // Real bytes, no readable footer: the shape a truncated upload leaves behind. + write(fileIO, tablePath, PARTITION_DIR + "/data-1.parquet", 16); + + PartitionStatistics measured = + new FormatTablePartitionStatsCollector(parquetTable(fileIO, tablePath), true, 1) + .collect(Collections.singletonList(spec("2025", "10"))) + .get(0); + + // One unreadable footer poisons the whole partition row count: a sum missing a file, + // reported as exact, is worse than no number at all. The other fields stay measured. + assertThat(PartitionStatistics.isKnown(measured.recordCount())).isFalse(); + assertThat(measured.fileCount()).isEqualTo(2); + assertThat(measured.fileSizeInBytes()).isPositive(); + } + + @Test + void testRowCountsStayWithTheirPartitionWhenMeasuredInParallel() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + writeParquet(fileIO, tablePath, PARTITION_DIR + "/data-0.parquet", 3); + writeParquet(fileIO, tablePath, "year=2025/month=11/data-0.parquet", 5); + + List measured = + new FormatTablePartitionStatsCollector(parquetTable(fileIO, tablePath), true, 2) + .collect(Arrays.asList(spec("2025", "10"), spec("2025", "11"))); + + // The listing and the footer reads run on a shared pool; each count still lands on the + // partition its file came from. + assertThat(measured.get(0).recordCount()).isEqualTo(3); + assertThat(measured.get(1).recordCount()).isEqualTo(5); + } + + @Test + void testCountingAnEmptyPartitionMeasuresAnExactZero() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + // Registered and never written to, holding only a staging tree, and gone from storage. + fileIO.mkdirs(new Path(tablePath, PARTITION_DIR)); + write(fileIO, tablePath, "year=2025/month=11/_temporary/attempt/part-0.parquet", 16); + + List measured = + new FormatTablePartitionStatsCollector(parquetTable(fileIO, tablePath), true, 1) + .collect( + Arrays.asList( + spec("2025", "10"), + spec("2025", "11"), + spec("2025", "12"))); + + // A measurement that counts rows has learned that the partition holds none, so it writes + // an exact zero; an unknown would leave the last count standing. + assertThat(measured) + .allSatisfy( + statistics -> { + assertThat(statistics.recordCount()).isZero(); + assertThat(statistics.fileCount()).isZero(); + assertThat( + PartitionStatistics.isKnown( + statistics.lastFileCreationTime())) + .isFalse(); + }); + } + + @Test + void testCsvKeepsAnUnknownRowCountEvenWhenAsked() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + write(fileIO, tablePath, PARTITION_DIR + "/data-0.csv", 100); + + PartitionStatistics measured = measure(fileIO, tablePath, true); + + assertThat(measured.fileCount()).isEqualTo(1); + // CSV carries no footer: an unknown row count beats a guessed one. + assertThat(PartitionStatistics.isKnown(measured.recordCount())).isFalse(); + } + private PartitionStatistics measure(FileIO fileIO, Path tablePath) { - return new FormatTablePartitionStatsCollector(table(fileIO, tablePath), 1) + return measure(fileIO, tablePath, false); + } + + private PartitionStatistics measure(FileIO fileIO, Path tablePath, boolean withRecordCount) { + return new FormatTablePartitionStatsCollector(table(fileIO, tablePath), withRecordCount, 1) .collect(Collections.singletonList(spec("2025", "10"))) .get(0); } @@ -296,6 +425,29 @@ private FormatTable table(FileIO fileIO, Path tablePath) { return table(fileIO, tablePath, FormatTable.Format.CSV, "csv"); } + private FormatTable parquetTable(FileIO fileIO, Path tablePath) { + return table(fileIO, tablePath, FormatTable.Format.PARQUET, "parquet"); + } + + private static void writeParquet(FileIO fileIO, Path tablePath, String relativePath, int rows) + throws Exception { + Map options = new HashMap<>(); + options.put(CoreOptions.FILE_FORMAT.key(), "parquet"); + FormatWriterFactory factory = + FileFormat.fileFormat(new CoreOptions(options)) + .createWriterFactory( + RowType.builder().field("id", DataTypes.INT()).build()); + Path path = new Path(tablePath, relativePath); + fileIO.mkdirs(path.getParent()); + try (PositionOutputStream out = fileIO.newOutputStream(path, false)) { + FormatWriter writer = factory.create(out, "zstd"); + for (int i = 0; i < rows; i++) { + writer.addElement(GenericRow.of(i)); + } + writer.close(); + } + } + private FormatTable table(FileIO fileIO, Path tablePath, Map options) { return table(fileIO, tablePath, FormatTable.Format.CSV, options); } diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java index 4c0ec743ff4f..4dd9329d1c4c 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java @@ -169,8 +169,9 @@ public class SparkConnectorOptions { .intType() .defaultValue(8) .withDescription( - "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."); + "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."); public static final ConfigOption SOURCE_SPLIT_TARGET_SIZE_WITH_COLUMN_PRUNING = key("source.split.target-size-with-column-pruning") diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala index 1ecb417abe4d..40276aca1c7f 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala @@ -23,7 +23,8 @@ import org.apache.paimon.spark.SparkTable import org.apache.paimon.spark.catalyst.Compatibility import org.apache.paimon.spark.catalyst.analysis.PaimonRelation.isPaimonTable import org.apache.paimon.spark.catalyst.plans.logical.{PaimonDropPartitions, PaimonHiveDynamicPartitionQuery} -import org.apache.paimon.spark.commands.{PaimonAnalyzeTableColumnCommand, PaimonDynamicPartitionOverwriteCommand, PaimonShowColumnsCommand, SchemaEvolutionHelper} +import org.apache.paimon.spark.commands.{PaimonAnalyzeFormatTablePartitionsCommand, PaimonAnalyzeTableColumnCommand, PaimonDynamicPartitionOverwriteCommand, PaimonShowColumnsCommand, SchemaEvolutionHelper} +import org.apache.paimon.spark.format.PaimonFormatTable import org.apache.paimon.spark.util.OptionUtils import org.apache.paimon.table.FileStoreTable @@ -282,6 +283,15 @@ case class PaimonPostHocResolutionRules(session: SparkSession) extends Rule[Logi } withoutHiveDynamicPartitionMarkers match { + // Spark rejects ANALYZE TABLE for every v2 table, so intercept before it does. Unlike a + // Paimon table, a Format Table has no snapshot to carry statistics and no column statistics + // to compute: analyzing it measures its partitions, for which PARTITION(...) and NOSCAN both + // mean something. Tables using filesystem partition discovery have no catalog to write to + // and fall through to the upstream rejection. + case a @ AnalyzeTable(ResolvedTable(_, _, table: PaimonFormatTable, _), partitionSpec, noScan) + if a.resolved && table.hasCatalogManagedPartitions => + PaimonAnalyzeFormatTablePartitionsCommand(table, partitionSpec, noScan) + case a @ AnalyzeTable( ResolvedTable(catalog, identifier, table: SparkTable, _), partitionSpec, diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeFormatTablePartitionsCommand.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeFormatTablePartitionsCommand.scala new file mode 100644 index 000000000000..1157b0342529 --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeFormatTablePartitionsCommand.scala @@ -0,0 +1,131 @@ +/* + * 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.spark.commands + +import org.apache.paimon.spark.catalyst.analysis.PaimonResolvePartitionSpec +import org.apache.paimon.spark.format.PaimonFormatTable +import org.apache.paimon.spark.leafnode.PaimonLeafRunnableCommand +import org.apache.paimon.spark.util.OptionUtils +import org.apache.paimon.table.format.FormatTablePartitionStatsCollector + +import org.apache.spark.sql.{Row, SparkSession} +import org.apache.spark.sql.PaimonUtils.normalizePartitionSpec +import org.apache.spark.sql.catalyst.analysis.NoSuchPartitionException +import org.apache.spark.sql.catalyst.expressions.GenericInternalRow +import org.apache.spark.sql.types.{StringType, StructField, StructType} +import org.apache.spark.unsafe.types.UTF8String + +import java.util.{Map => JMap} + +import scala.collection.JavaConverters._ +import scala.collection.immutable.ListMap + +/** + * Recomputes the catalog statistics of a Format Table with catalog-managed partitions, backing + * `ANALYZE TABLE t [PARTITION(...)] COMPUTE STATISTICS [NOSCAN]`. + * + * The partitions are measured from storage and each measured field replaces what the catalog holds, + * so this is how a table catches up with writers the catalog never saw. `NOSCAN` stops at what a + * directory listing gives — file count, byte size, last file creation time — while a full ANALYZE + * also reads each file footer for its row count. A format that carries no footer (CSV, TEXT, JSON) + * leaves the row count as it was rather than guessing one. + * + * Analyzing is not a way to add or remove partitions: it measures the ones registered at the time + * of the listing and re-registers exactly those. There is no lock between the listing and the + * write, so a partition dropped concurrently can be re-registered with its last measurement — the + * same last-writer-wins window every lock-free partition operation on these tables has. A + * `PARTITION(...)` clause selects the partitions whose leading values it fixes. + */ +case class PaimonAnalyzeFormatTablePartitionsCommand( + v2Table: PaimonFormatTable, + partitionSpec: Map[String, Option[String]], + noScan: Boolean) + extends PaimonLeafRunnableCommand { + + override def run(sparkSession: SparkSession): Seq[Row] = { + val prefix = leadingPrefix(sparkSession) + val partitions = v2Table.partitionManager + .listPartitions(prefix.asJava, null) + .asScala + .map(_.spec()) + .toList + + if (partitions.isEmpty && prefix.nonEmpty) { + throw new NoSuchPartitionException( + v2Table.name(), + new GenericInternalRow(prefix.values.map(UTF8String.fromString(_): Any).toArray), + StructType(prefix.keys.map(StructField(_, StringType)).toSeq) + ) + } + + if (partitions.nonEmpty) { + val collector = new FormatTablePartitionStatsCollector( + v2Table.table, + !noScan, + OptionUtils.formatTableStatisticsParallelism()) + val statistics = collector.collect(partitions.asJava) + v2Table.partitionManager + .createPartitions(partitions.asJava, true, statistics, true) + } + Seq.empty[Row] + } + + /** + * The values the `PARTITION(...)` clause fixes, as a leading prefix of the partition keys — the + * shape the catalog can select on. + * + * This follows what Spark does with the same clause on a metastore table: names are resolved by + * the same helper its own commands use, a column named without a value means every value of it, + * and the columns that do carry a value have to be a leading run. `PARTITION (dt = 'x', hour)` + * therefore selects every hour of that day and `PARTITION (dt, hour)` selects everything, while + * `PARTITION (hour = '00')` is rejected: the catalog cannot select on a non-leading key, and + * quietly widening it would measure more partitions than were asked for. + */ + private def leadingPrefix(sparkSession: SparkSession): Map[String, String] = { + if (partitionSpec.isEmpty) { + return Map.empty + } + val partitionKeys = v2Table.table.partitionKeys().asScala.toSeq + val normalized = normalizePartitionSpec( + partitionSpec, + v2Table.partitionSchema, + v2Table.name(), + sparkSession.sessionState.conf.resolver) + val valueByKey = partitionKeys.map(key => key -> normalized.get(key).flatten) + val prefix = valueByKey.takeWhile(_._2.isDefined) + if (valueByKey.drop(prefix.size).exists(_._2.isDefined)) { + throw new IllegalArgumentException( + s"ANALYZE TABLE ${v2Table.name()} PARTITION must give values for a leading run of its " + + s"partition columns ${partitionKeys.mkString("[", ", ", "]")}, but got values for " + + valueByKey.filter(_._2.isDefined).map(_._1).mkString("[", ", ", "]")) + } + if (prefix.isEmpty) { + return Map.empty + } + // The catalog holds the values the way Paimon writes them, so what the parser handed over is + // read as the partition column type first: PARTITION (p = '01') selects the INT partition + // registered as 1, and a null value selects the default partition. + val names = prefix.map { case (key, _) => key } + val ident = PaimonResolvePartitionSpec.convertToPartIdent( + prefix.map { case (key, value) => key -> value.get }.toMap, + names.map(v2Table.partitionSchema.apply)) + // Kept in partition-key order, so a message built from it reads in that order too. + ListMap(v2Table.toCatalogPartition(ident, names).asScala.toSeq: _*) + } +} diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/PaimonFormatTable.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/PaimonFormatTable.scala index 131fc4e2b360..fea982207dd4 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/PaimonFormatTable.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/PaimonFormatTable.scala @@ -179,6 +179,15 @@ case class PaimonFormatTable(table: FormatTable) } } + /** + * The catalog spec of a resolved partition identifier, with each value written the way Paimon + * writes it into a partition directory - a null becomes the default partition name. + */ + private[spark] def toCatalogPartition( + ident: InternalRow, + partitionNames: Seq[String]): JMap[String, String] = + toPaimonPartition(ident, partitionNames) + /** * Resolves, with a single catalog list-by-names lookup, which of the given complete partition * specs are registered. The result is aligned with the input arrays. diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionAnalyzeTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionAnalyzeTest.scala new file mode 100644 index 000000000000..89d57134f89c --- /dev/null +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionAnalyzeTest.scala @@ -0,0 +1,504 @@ +/* + * 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.spark.sql + +import org.apache.paimon.catalog.Identifier +import org.apache.paimon.fs.Path +import org.apache.paimon.partition.{Partition, PartitionStatistics} +import org.apache.paimon.spark.PaimonSparkTestWithRestCatalogBase +import org.apache.paimon.table.FormatTable + +import org.apache.spark.sql.catalyst.analysis.NoSuchPartitionException + +import java.util.Locale + +import scala.collection.JavaConverters._ + +/** + * `ANALYZE TABLE ... PARTITION(...) COMPUTE STATISTICS [NOSCAN]` on a Format Table with + * catalog-managed partitions, held against the semantics Spark and Hive give the same statement on + * a Hive metastore table. + * + * The reference behaviour is what Spark's own `StatisticsSuite` and its `AlterTable*Partition` + * command suites pin: a partition column named without a value means every value of it, a spec + * naming a partition that does not exist is an error rather than a no-op, and partition column + * names resolve the way the rest of Spark resolves identifiers. + */ +class CatalogManagedPartitionAnalyzeTest extends PaimonSparkTestWithRestCatalogBase { + + test("ANALYZE with a full partition spec measures only that partition") { + val tableName = "analyze_full_spec" + withTable(tableName) { + createTable(tableName) + writeCsvPartition(tableName, "20260101", "00", 1) + writeCsvPartition(tableName, "20260101", "01", 2) + repair(tableName) + + sql( + s"ANALYZE TABLE ${qualified(tableName)} PARTITION (dt = '20260101', hour = '00') " + + s"COMPUTE STATISTICS NOSCAN").collect() + + assert(statisticsOf(tableName, "20260101", "00").fileCount() == 1L) + assert(!PartitionStatistics.isKnown(statisticsOf(tableName, "20260101", "01").fileCount())) + } + } + + test("ANALYZE with a leading prefix measures every partition under it") { + val tableName = "analyze_prefix" + withTable(tableName) { + createTable(tableName) + writeCsvPartition(tableName, "20260101", "00", 1) + writeCsvPartition(tableName, "20260101", "01", 2) + writeCsvPartition(tableName, "20260102", "00", 3) + repair(tableName) + + sql( + s"ANALYZE TABLE ${qualified(tableName)} PARTITION (dt = '20260101') " + + s"COMPUTE STATISTICS NOSCAN").collect() + + assert(statisticsOf(tableName, "20260101", "00").fileCount() == 1L) + assert(statisticsOf(tableName, "20260101", "01").fileCount() == 1L) + // The sibling day is outside the prefix and keeps whatever it had. + assert(!PartitionStatistics.isKnown(statisticsOf(tableName, "20260102", "00").fileCount())) + } + } + + test("ANALYZE naming every partition column without a value measures every partition") { + val tableName = "analyze_all_columns" + withTable(tableName) { + createTable(tableName) + writeCsvPartition(tableName, "20260101", "00", 1) + writeCsvPartition(tableName, "20260102", "01", 2) + repair(tableName) + + // Spark and Hive read `PARTITION (dt, hour)` as every value of both columns. + sql(s"ANALYZE TABLE ${qualified(tableName)} PARTITION (dt, hour) COMPUTE STATISTICS NOSCAN") + .collect() + + assert(statisticsOf(tableName, "20260101", "00").fileCount() == 1L) + assert(statisticsOf(tableName, "20260102", "01").fileCount() == 1L) + } + } + + test("ANALYZE naming a trailing column without a value measures the set under the prefix") { + val tableName = "analyze_partial_values" + withTable(tableName) { + createTable(tableName) + writeCsvPartition(tableName, "20260101", "00", 1) + writeCsvPartition(tableName, "20260101", "01", 2) + writeCsvPartition(tableName, "20260102", "00", 3) + repair(tableName) + + // Spark and Hive read `PARTITION (dt = 'x', hour)` as every hour of that day. + sql( + s"ANALYZE TABLE ${qualified(tableName)} PARTITION (dt = '20260101', hour) " + + s"COMPUTE STATISTICS NOSCAN").collect() + + assert(statisticsOf(tableName, "20260101", "00").fileCount() == 1L) + assert(statisticsOf(tableName, "20260101", "01").fileCount() == 1L) + assert(!PartitionStatistics.isKnown(statisticsOf(tableName, "20260102", "00").fileCount())) + } + } + + test("ANALYZE of a partition that does not exist fails instead of measuring nothing") { + val tableName = "analyze_missing_partition" + withTable(tableName) { + createTable(tableName) + writeCsvPartition(tableName, "20260101", "00", 1) + repair(tableName) + + // Succeeding silently tells the caller a partition was measured when none was. + val error = intercept[Exception] { + sql( + s"ANALYZE TABLE ${qualified(tableName)} PARTITION (dt = '20991231') " + + s"COMPUTE STATISTICS NOSCAN").collect() + } + // The same error Spark reports for a partition it cannot find on any other table. + assert(causeMessages(error).contains("20991231"), causeMessages(error)) + assert( + error.isInstanceOf[NoSuchPartitionException] || + error.getCause.isInstanceOf[NoSuchPartitionException], + error) + } + } + + test("ANALYZE of a non-leading partition column is rejected instead of widened") { + val tableName = "analyze_non_leading" + withTable(tableName) { + createTable(tableName) + writeCsvPartition(tableName, "20260101", "00", 1) + writeCsvPartition(tableName, "20260102", "00", 2) + repair(tableName) + + // The catalog selects on a leading run, so the only way to serve this spec is to measure + // every day that has an hour 00 — more partitions than were asked about. + val error = intercept[Exception] { + sql( + s"ANALYZE TABLE ${qualified(tableName)} PARTITION (hour = '00') " + + s"COMPUTE STATISTICS NOSCAN").collect() + } + assert( + causeMessages(error).contains("leading run of its partition columns [dt, hour]"), + causeMessages(error)) + // Rejected means nothing was measured, in either of the two days the widening would reach. + assert(!PartitionStatistics.isKnown(statisticsOf(tableName, "20260101", "00").fileCount())) + assert(!PartitionStatistics.isKnown(statisticsOf(tableName, "20260102", "00").fileCount())) + } + } + + test("ANALYZE of a column that is not a partition column is rejected") { + val tableName = "analyze_not_a_partition_column" + withTable(tableName) { + createTable(tableName) + writeCsvPartition(tableName, "20260101", "00", 1) + repair(tableName) + + // `payload` is a column of the table but not one the catalog partitions on. + val error = intercept[Exception] { + sql( + s"ANALYZE TABLE ${qualified(tableName)} PARTITION (payload = 'a') " + + s"COMPUTE STATISTICS NOSCAN").collect() + } + assert(causeMessages(error).contains("payload"), causeMessages(error)) + // Dropping the name from the spec instead would measure the whole table. + assert(!PartitionStatistics.isKnown(statisticsOf(tableName, "20260101", "00").fileCount())) + } + } + + test("ANALYZE resolves partition column names the way the rest of Spark resolves them") { + val tableName = "analyze_case" + withTable(tableName) { + createTable(tableName) + writeCsvPartition(tableName, "20260101", "00", 1) + repair(tableName) + + sql( + s"ANALYZE TABLE ${qualified(tableName)} PARTITION (DT = '20260101') " + + s"COMPUTE STATISTICS NOSCAN").collect() + + assert(statisticsOf(tableName, "20260101", "00").fileCount() == 1L) + + // And a case-sensitive session resolves it the way the rest of that session does. + withSQLConf("spark.sql.caseSensitive" -> "true") { + val error = intercept[Exception] { + sql( + s"ANALYZE TABLE ${qualified(tableName)} PARTITION (DT = '20260101') " + + s"COMPUTE STATISTICS NOSCAN").collect() + } + assert(causeMessages(error).contains("DT"), causeMessages(error)) + } + } + } + + test("ANALYZE reads a partition value as the type of its partition column") { + val tableName = "analyze_typed_partition" + withTable(tableName) { + sql(s"""CREATE TABLE $tableName (id INT, p INT) + |USING CSV + |PARTITIONED BY (p) + |TBLPROPERTIES ( + | 'format-table.implementation' = 'paimon', + | 'metastore.partitioned-table' = 'true') + |""".stripMargin) + val table = formatTable(tableName) + val partitionPath = new Path(table.location(), "p=1") + table.fileIO().mkdirs(partitionPath) + table.fileIO().writeFile(new Path(partitionPath, "part-00001.csv"), "1\n", false) + repair(tableName) + + // The catalog holds the value the way Paimon writes it, so '01' names the partition it + // registered as 1 rather than one that does not exist. + sql(s"ANALYZE TABLE ${qualified(tableName)} PARTITION (p = '01') COMPUTE STATISTICS NOSCAN") + .collect() + + assert(partitionOf(tableName, "p", "1").fileCount() == 1L) + } + } + + test("ANALYZE of a table with no registered partitions measures nothing") { + val tableName = "analyze_no_registered_partitions" + withTable(tableName) { + createTable(tableName) + // The other half of the guard that fails a PARTITION spec matching nothing: with no spec + // there is nothing to have missed, so measuring nothing is the answer, not an error. + writeCsvPartition(tableName, "20260101", "00", 1) + + sql(s"ANALYZE TABLE ${qualified(tableName)} COMPUTE STATISTICS NOSCAN").collect() + + assert(registeredPartitions(tableName).isEmpty) + } + } + + test("ANALYZE ... FOR COLUMNS is rejected for a format table") { + val tableName = "analyze_for_columns" + withTable(tableName) { + createTable(tableName) + writeCsvPartition(tableName, "20260101", "00", 1) + repair(tableName) + + // A Format Table has nowhere to keep column statistics, so only the partition measurement + // is intercepted and the column form keeps Spark's own rejection. + intercept[Exception] { + sql(s"ANALYZE TABLE ${qualified(tableName)} COMPUTE STATISTICS FOR ALL COLUMNS").collect() + } + intercept[Exception] { + sql(s"ANALYZE TABLE ${qualified(tableName)} COMPUTE STATISTICS FOR COLUMNS payload") + .collect() + } + assert(!PartitionStatistics.isKnown(statisticsOf(tableName, "20260101", "00").fileCount())) + } + } + + test("NOSCAN keeps a row count that is already known") { + val tableName = "analyze_noscan_keeps" + withTable(tableName) { + // A full ANALYZE is the way this suite can put an exact row count in the catalog, so the + // table is parquet; what the NOSCAN below must not do is erase it, however it was learned. + sql(s"""CREATE TABLE $tableName (id INT, payload STRING, dt STRING, hour STRING) + |USING PARQUET + |PARTITIONED BY (dt, hour) + |TBLPROPERTIES ( + | 'format-table.implementation' = 'paimon', + | 'metastore.partitioned-table' = 'true') + |""".stripMargin) + sql(s"INSERT INTO ${qualified(tableName)} VALUES (1, 'a', '20260101', '00')") + sql(s"ANALYZE TABLE ${qualified(tableName)} COMPUTE STATISTICS").collect() + val scanned = statisticsOf(tableName, "20260101", "00") + assert(scanned.recordCount() == 1L, scanned.toString) + + // A listing cannot count rows, but it also learned nothing that contradicts the count that + // is already there. Hive keeps numRows across a NOSCAN for exactly this reason. + sql(s"ANALYZE TABLE ${qualified(tableName)} COMPUTE STATISTICS NOSCAN").collect() + + val afterNoScan = statisticsOf(tableName, "20260101", "00") + assert(afterNoScan.fileCount() == 1L, afterNoScan.toString) + assert(afterNoScan.recordCount() == 1L, afterNoScan.toString) + } + } + + test("a full ANALYZE reads the row count a NOSCAN cannot") { + val tableName = "analyze_footers" + withTable(tableName) { + // Parquet carries a row count in its footer. CSV, which the rest of this suite uses, carries + // none, so it is the format that cannot tell a full ANALYZE apart from a NOSCAN. + sql(s"""CREATE TABLE $tableName (id INT, payload STRING, dt STRING, hour STRING) + |USING PARQUET + |PARTITIONED BY (dt, hour) + |TBLPROPERTIES ( + | 'format-table.implementation' = 'paimon', + | 'metastore.partitioned-table' = 'true') + |""".stripMargin) + sql(s"""INSERT INTO ${qualified(tableName)} + |VALUES (1, 'a', '20260101', '00'), (2, 'b', '20260101', '00') + |""".stripMargin) + // A commit reports the rows it wrote, so a partition it registered is already measured. The + // two partitions below hold the same parquet files copied in from outside and registered by + // a repair, which reports nothing, so their row count is nobody's measurement yet. + copyPartitionFiles(tableName, "20260101", "20260102") + copyPartitionFiles(tableName, "20260101", "20260103") + repair(tableName) + assert(!PartitionStatistics.isKnown(statisticsOf(tableName, "20260102", "00").recordCount())) + + sql(s"ANALYZE TABLE ${qualified(tableName)} PARTITION (dt = '20260102') COMPUTE STATISTICS") + .collect() + + // No NOSCAN, so the footers were read and the row count is exact. + val scanned = statisticsOf(tableName, "20260102", "00") + assert(scanned.recordCount() == 2L, scanned.toString) + assert(scanned.fileCount() >= 1L, scanned.toString) + + sql( + s"ANALYZE TABLE ${qualified(tableName)} PARTITION (dt = '20260103') " + + s"COMPUTE STATISTICS NOSCAN").collect() + + // NOSCAN measured from the listing alone: the file numbers are there, and the row count is + // still nobody's measurement even though this format could have given one. + val listed = statisticsOf(tableName, "20260103", "00") + assert(listed.fileCount() >= 1L, listed.toString) + assert(listed.fileSizeInBytes() > 0L, listed.toString) + assert(!PartitionStatistics.isKnown(listed.recordCount()), listed.toString) + } + } + + test("ANALYZE run twice reports the same measurement rather than accumulating") { + val tableName = "analyze_idempotent" + withTable(tableName) { + createTable(tableName) + writeCsvPartition(tableName, "20260101", "00", 1) + repair(tableName) + + sql(s"ANALYZE TABLE ${qualified(tableName)} COMPUTE STATISTICS NOSCAN").collect() + val once = statisticsOf(tableName, "20260101", "00") + sql(s"ANALYZE TABLE ${qualified(tableName)} COMPUTE STATISTICS NOSCAN").collect() + val twice = statisticsOf(tableName, "20260101", "00") + + // Anchored, so two runs that both measured nothing cannot pass as two equal measurements. + assert(once.fileCount() == 1L, once.toString) + assert(once.fileSizeInBytes() > 0L, once.toString) + assert(twice.fileCount() == once.fileCount(), s"$once then $twice") + assert(twice.fileSizeInBytes() == once.fileSizeInBytes(), s"$once then $twice") + } + } + + test("ANALYZE does not count files a committer left staged in the partition") { + val tableName = "analyze_staging" + withTable(tableName) { + createTable(tableName) + val partitionPath = writeCsvPartition(tableName, "20260101", "00", 1) + val table = formatTable(tableName) + // What a magic committer leaves behind: a data file name under a staging directory. + val staged = + new Path(new Path(partitionPath, "__magic_job-1/tasks/attempt_1/__base"), "part-9.csv") + table.fileIO().mkdirs(staged.getParent) + table.fileIO().writeFile(staged, "9,staged\n", false) + repair(tableName) + + sql(s"ANALYZE TABLE ${qualified(tableName)} COMPUTE STATISTICS NOSCAN").collect() + + val measured = statisticsOf(tableName, "20260101", "00") + // The reader returns one row from one file; a measurement claiming two is a number no query + // can reproduce. + assert(measured.fileCount() == 1L, measured.toString) + assert(sql(s"SELECT COUNT(*) FROM ${qualified(tableName)}").collect()(0).getLong(0) == 1L) + } + } + + test("ANALYZE measures registered partitions and never changes which exist") { + val tableName = "analyze_partition_set" + withTable(tableName) { + createTable(tableName) + writeCsvPartition(tableName, "20260101", "00", 1) + writeCsvPartition(tableName, "20260102", "00", 2) + repair(tableName) + + sql( + s"ANALYZE TABLE ${qualified(tableName)} PARTITION (dt = '20260101') " + + s"COMPUTE STATISTICS NOSCAN").collect() + + val scoped = statisticsOf(tableName, "20260101", "00") + assert(scoped.fileCount() == 1L, scoped.toString) + assert(scoped.fileSizeInBytes() > 0L, scoped.toString) + // A PARTITION clause scopes the measurement; the sibling keeps whatever it had. + val sibling = statisticsOf(tableName, "20260102", "00") + assert(!PartitionStatistics.isKnown(sibling.fileCount()), sibling.toString) + + sql(s"ANALYZE TABLE ${qualified(tableName)} COMPUTE STATISTICS NOSCAN").collect() + assert(statisticsOf(tableName, "20260102", "00").fileCount() == 1L) + // Analyzing measures partitions, it does not decide which ones exist. + val expected = Set("dt=20260101/hour=00", "dt=20260102/hour=00") + assert(registeredPartitions(tableName) == expected) + assert( + sql(s"SHOW PARTITIONS ${qualified(tableName)}").collect().map(_.getString(0)).toSet == + expected) + } + } + + test("ANALYZE is rejected for a format table discovering partitions from the filesystem") { + val tableName = "analyze_filesystem_partitions" + withTable(tableName) { + sql(s"""CREATE TABLE $tableName (id INT, payload STRING, dt STRING, hour STRING) + |USING CSV + |PARTITIONED BY (dt, hour) + |TBLPROPERTIES ( + | 'format-table.implementation' = 'paimon', + | 'metastore.partitioned-table' = 'false') + |""".stripMargin) + + // There is no catalog to write a measurement to, so the table is not intercepted at all and + // keeps Spark's own rejection rather than quietly measuring nothing. + val error = intercept[Exception] { + sql(s"ANALYZE TABLE ${qualified(tableName)} COMPUTE STATISTICS NOSCAN").collect() + } + val messages = causeMessages(error) + assert(messages.contains("ANALYZE TABLE"), messages) + assert(messages.toLowerCase(Locale.ROOT).contains("not supported"), messages) + } + } + + private def qualified(tableName: String): String = s"paimon.$dbName0.$tableName" + + private def createTable(tableName: String): Unit = { + sql(s"""CREATE TABLE $tableName (id INT, payload STRING, dt STRING, hour STRING) + |USING CSV + |PARTITIONED BY (dt, hour) + |TBLPROPERTIES ( + | 'format-table.implementation' = 'paimon', + | 'metastore.partitioned-table' = 'true') + |""".stripMargin) + } + + private def repair(tableName: String): Unit = + sql(s"MSCK REPAIR TABLE ${qualified(tableName)}").collect() + + private def formatTable(tableName: String): FormatTable = + paimonCatalog.getTable(Identifier.create(dbName0, tableName)).asInstanceOf[FormatTable] + + private def writeCsvPartition(tableName: String, dt: String, hour: String, id: Int): Path = { + val table = formatTable(tableName) + val partitionPath = new Path(table.location(), s"dt=$dt/hour=$hour") + table.fileIO().mkdirs(partitionPath) + table + .fileIO() + .writeFile(new Path(partitionPath, f"part-$id%05d.csv"), s"$id,payload-$id\n", false) + partitionPath + } + + /** The same files under another partition value: written by nobody this catalog heard from. */ + private def copyPartitionFiles(tableName: String, sourceDt: String, targetDt: String): Unit = { + val table = formatTable(tableName) + val source = new Path(table.location(), s"dt=$sourceDt/hour=00") + val target = new Path(table.location(), s"dt=$targetDt/hour=00") + table.fileIO().mkdirs(target) + table + .fileIO() + .listStatus(source) + .filter(!_.isDir) + .foreach( + status => + table.fileIO().copyFile(status.getPath, new Path(target, status.getPath.getName), false)) + } + + private def registeredPartitions(tableName: String): Set[String] = + paimonCatalog + .listPartitions(Identifier.create(dbName0, tableName)) + .asScala + .map(partition => s"dt=${partition.spec().get("dt")}/hour=${partition.spec().get("hour")}") + .toSet + + private def statisticsOf(tableName: String, dt: String, hour: String): Partition = + paimonCatalog + .listPartitions(Identifier.create(dbName0, tableName)) + .asScala + .find(p => p.spec().get("dt") == dt && p.spec().get("hour") == hour) + .getOrElse(fail(s"partition dt=$dt/hour=$hour of $tableName is not registered")) + + private def partitionOf(tableName: String, key: String, value: String): Partition = + paimonCatalog + .listPartitions(Identifier.create(dbName0, tableName)) + .asScala + .find(_.spec().get(key) == value) + .getOrElse(fail(s"partition $key=$value of $tableName is not registered")) + + private def causeMessages(error: Throwable): String = + Iterator + .iterate(error)(_.getCause) + .takeWhile(_ != null) + .map(e => String.valueOf(e.getMessage)) + .mkString(" | ") +} From 082d73d0e8e5d3237de35e0293754ecd075bbf36 Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Fri, 21 Aug 2026 18:20:33 +0800 Subject: [PATCH 2/2] [spark] Run full ANALYZE footer reads on the executors A full ANALYZE opens the footer of every file, so reading them on the driver makes the statement driver-bound on a table with many files and leaves the cluster idle, while the driver holds a listing entry for every one of them. The partitions now go through a Spark job: each task builds one footer reader and returns what its partitions add up to, so the driver only ever holds one measurement per partition. format-table.statistics.parallelism bounds how many requests are in flight: it caps the tasks, and what is left of it caps the files each task reads at once. NOSCAN needs only a listing and keeps taking it locally. --- docs/docs/spark/sql-ddl.md | 3 +- ...nAnalyzeFormatTablePartitionsCommand.scala | 59 ++++++++++++++++--- 2 files changed, 53 insertions(+), 9 deletions(-) diff --git a/docs/docs/spark/sql-ddl.md b/docs/docs/spark/sql-ddl.md index 336705206a58..e4ef5c707d82 100644 --- a/docs/docs/spark/sql-ddl.md +++ b/docs/docs/spark/sql-ddl.md @@ -238,7 +238,8 @@ use `MSCK REPAIR TABLE` for that. 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 `NOSCAN` is the cheaper of the two. +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)`, diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeFormatTablePartitionsCommand.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeFormatTablePartitionsCommand.scala index 1157b0342529..06aee2ffa39d 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeFormatTablePartitionsCommand.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeFormatTablePartitionsCommand.scala @@ -18,6 +18,7 @@ package org.apache.paimon.spark.commands +import org.apache.paimon.partition.PartitionStatistics import org.apache.paimon.spark.catalyst.analysis.PaimonResolvePartitionSpec import org.apache.paimon.spark.format.PaimonFormatTable import org.apache.paimon.spark.leafnode.PaimonLeafRunnableCommand @@ -31,7 +32,7 @@ import org.apache.spark.sql.catalyst.expressions.GenericInternalRow import org.apache.spark.sql.types.{StringType, StructField, StructType} import org.apache.spark.unsafe.types.UTF8String -import java.util.{Map => JMap} +import java.util.{List => JList, Map => JMap} import scala.collection.JavaConverters._ import scala.collection.immutable.ListMap @@ -43,8 +44,9 @@ import scala.collection.immutable.ListMap * The partitions are measured from storage and each measured field replaces what the catalog holds, * so this is how a table catches up with writers the catalog never saw. `NOSCAN` stops at what a * directory listing gives — file count, byte size, last file creation time — while a full ANALYZE - * also reads each file footer for its row count. A format that carries no footer (CSV, TEXT, JSON) - * leaves the row count as it was rather than guessing one. + * also reads each file footer for its row count, on the executors, since that is one open per file. + * A format that carries no footer (CSV, TEXT, JSON) leaves the row count as it was rather than + * guessing one. * * Analyzing is not a way to add or remove partitions: it measures the ones registered at the time * of the listing and re-registers exactly those. There is no lock between the listing and the @@ -75,17 +77,58 @@ case class PaimonAnalyzeFormatTablePartitionsCommand( } if (partitions.nonEmpty) { - val collector = new FormatTablePartitionStatsCollector( - v2Table.table, - !noScan, - OptionUtils.formatTableStatisticsParallelism()) - val statistics = collector.collect(partitions.asJava) + val parallelism = OptionUtils.formatTableStatisticsParallelism() + val statistics = + if (noScan) { + // One listing request per partition is all NOSCAN reports, cheap enough to take here. + new FormatTablePartitionStatsCollector(v2Table.table, false, parallelism) + .collect(partitions.asJava) + } else { + measureOnExecutors(sparkSession, partitions, parallelism) + } v2Table.partitionManager .createPartitions(partitions.asJava, true, statistics, true) } Seq.empty[Row] } + /** + * Measures the partitions on the executors. A full ANALYZE opens the footer of every file, so + * reading them here would make the statement driver-bound on a table with many files and leave + * the cluster idle. Each task builds one footer reader and returns what its partitions add up to, + * so the driver only ever holds one measurement per partition. + * + * `format-table.statistics.parallelism` bounds how many requests are in flight: it caps the + * tasks, and what is left of it caps the files each task reads at once. + */ + private def measureOnExecutors( + sparkSession: SparkSession, + partitions: List[JMap[String, String]], + parallelism: Int): JList[PartitionStatistics] = { + val tasks = math.min(parallelism, partitions.size) + val perTask = math.max(1, parallelism / tasks) + // The table, not this command: a Spark table cannot be shipped to an executor. + val table = v2Table.table + val measured = sparkSession.sparkContext + .parallelize(partitions.zipWithIndex, tasks) + .mapPartitions { + batch => + val work = batch.toSeq + if (work.isEmpty) { + Iterator.empty + } else { + val collector = new FormatTablePartitionStatsCollector(table, true, perTask) + work + .map(_._2) + .iterator + .zip(collector.collect(work.map(_._1).asJava).asScala.iterator) + } + } + .collect() + .toMap + partitions.indices.map(measured).asJava + } + /** * The values the `PARTITION(...)` clause fixes, as a leading prefix of the partition keys — the * shape the catalog can select on.