From 6a78c95c002c8db50faeba9d8adf29b99ab9cbce 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:00:19 +0800 Subject: [PATCH 1/2] [core][spark] Measure format table partitions in MSCK REPAIR TABLE MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A commit reports what it wrote. Nothing reports what is already there, and for a format table plenty is: partitions written by something that is not Paimon, files deleted out of band, an increment redelivered and counted twice. MSCK REPAIR TABLE is already the command that reconciles the partition set against the directories, so it is the natural place to reconcile the numbers too. Off by default, behind spark.paimon.format-table.repair.collect-statistics, because measuring changes what a repair costs: the plain diff lists partition directories, and measuring lists the files inside every one of them. That is a different order of magnitude on a table with many partitions, and a repair should not silently become that. When it is on, spark.paimon.format-table.statistics.parallelism caps how many partitions are measured at once, at 8: listing one is a round trip the driver spends waiting on, and the cap keeps a table with many partitions from turning that wait into a burst of requests. When on it measures every partition that ends up registered with a directory behind it, not only the ones it just added — the stale numbers of partitions written outside Paimon are exactly what a repair exists to correct. Without ADD it stays inside the already-registered set, so measuring never registers a partition the command was not asked to. The collector reports what a reader would see. File count, byte size and last file creation time come from the listing. It stops there. The row count it leaves unknown: reading it means opening every file's footer, which no listing does and which the command that wants exact rows can pay for on its own. A listing failure aborts the whole collection rather than reporting what it managed to see, because a truncated listing is indistinguishable from a partition that lost files. A partition whose directory is gone measures as an exact zero, with no last file to date. Tests: FormatTablePartitionStatsCollectorTest covers staging trees, hidden files beside the data, a missing directory, the value-only layout and a value the directory name has to escape, alignment with the given specs, a spec missing a partition key, and a listing failure aborting both the serial and the parallel path. FormatTablePartitionRepairTest covers measuring every partition on disk, a repair without ADD registering nothing, and a listing failure leaving the catalog untouched. CatalogManagedPartitionMsckRepairTest covers the command end to end, with the option off and on. --- .../spark_connector_configuration.html | 12 + .../FormatTablePartitionStatsCollector.java | 195 ++++++++++ ...ormatTablePartitionStatsCollectorTest.java | 346 ++++++++++++++++++ .../paimon/spark/SparkConnectorOptions.java | 17 + .../format/FormatTablePartitionRepair.java | 77 +++- .../PaimonFormatTablePartitionDdlExec.scala | 14 +- .../paimon/spark/util/OptionUtils.scala | 8 + .../FormatTablePartitionRepairTest.java | 147 +++++++- ...atalogManagedPartitionMsckRepairTest.scala | 44 ++- 9 files changed, 841 insertions(+), 19 deletions(-) create mode 100644 paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionStatsCollector.java create mode 100644 paimon-core/src/test/java/org/apache/paimon/table/format/FormatTablePartitionStatsCollectorTest.java diff --git a/docs/generated/spark_connector_configuration.html b/docs/generated/spark_connector_configuration.html index 875b3d563994..71bb3861488a 100644 --- a/docs/generated/spark_connector_configuration.html +++ b/docs/generated/spark_connector_configuration.html @@ -26,6 +26,18 @@ + +
format-table.repair.collect-statistics
+ false + Boolean + Whether MSCK REPAIR TABLE on a Format Table also measures the partitions it finds. Off by default: measuring lists the files inside every partition, not only the partition directories. + + +
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. +
legacy-timestamp-mapping.enabled
false 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 new file mode 100644 index 000000000000..0f2323546b94 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionStatsCollector.java @@ -0,0 +1,195 @@ +/* + * 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.fs.FileIO; +import org.apache.paimon.fs.FileStatus; +import org.apache.paimon.fs.Path; +import org.apache.paimon.partition.PartitionStatistics; +import org.apache.paimon.table.FormatTable; +import org.apache.paimon.utils.PartitionPathUtils; +import org.apache.paimon.utils.ThreadPoolUtils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +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.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. + * + *

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 + * rather than walked. A listing failure aborts the whole collection: a truncated listing looks + * exactly like a partition that lost files. + * + *

The result is a whole-partition measurement, so it replaces rather than accumulates. It never + * decides that a partition should exist or stop existing; it measures the ones it is given. + */ +public class FormatTablePartitionStatsCollector { + + private static final Logger LOG = + LoggerFactory.getLogger(FormatTablePartitionStatsCollector.class); + + private final FormatTable table; + + private final boolean onlyValueInPath; + private final int parallelism; + + public FormatTablePartitionStatsCollector(FormatTable table, int parallelism) { + this.table = table; + this.onlyValueInPath = + new CoreOptions(table.options()).formatTablePartitionOnlyValueInPath(); + 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 collect(List> partitions) { + if (partitions.isEmpty()) { + return Collections.emptyList(); + } + int threads = Math.min(parallelism, partitions.size()); + if (threads == 1) { + List statistics = new ArrayList<>(partitions.size()); + for (Map partition : partitions) { + statistics.add(measure(partition)); + } + return statistics; + } + + ExecutorService executor = + ThreadPoolUtils.createCachedThreadPool(threads, "FORMAT-TABLE-STATS-THREAD-POOL"); + try { + List> futures = new ArrayList<>(partitions.size()); + for (Map partition : partitions) { + futures.add(executor.submit(() -> measure(partition))); + } + 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()); + } + } + return statistics; + } finally { + executor.shutdownNow(); + } + } + + private PartitionStatistics measure(Map partition) { + FileIO fileIO = table.fileIO(); + 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); + } catch (FileNotFoundException e) { + // A registered partition whose directory is gone reads as empty. + return empty(partition); + } 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; + 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, + fileSizeInBytes, + fileCount, + 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); + } + + private Path partitionPath(Map partition) { + LinkedHashMap 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); + } +} 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 new file mode 100644 index 000000000000..522288c1e018 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTablePartitionStatsCollectorTest.java @@ -0,0 +1,346 @@ +/* + * 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.catalog.Identifier; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.FileStatus; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.PositionOutputStream; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.partition.PartitionStatistics; +import org.apache.paimon.table.FormatTable; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowType; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Tests for what {@link FormatTablePartitionStatsCollector} measures, which is what {@code ANALYZE + * TABLE} and a measuring {@code MSCK REPAIR} write into the catalog. + * + *

The staging cases are the same ones the read path has to survive: a committer leaves trees + * such as {@code _temporary/}, {@code __magic_job-/} and {@code .hive-staging_*} inside the + * partition, and the files under them carry ordinary data file names. A measurement that counts + * them reports a partition that holds more than any reader will ever return. + */ +class FormatTablePartitionStatsCollectorTest { + + private static final Identifier TABLE = + Identifier.create("statistics_db", "statistics_format_table"); + private static final String PARTITION_DIR = "year=2025/month=10"; + + @TempDir java.nio.file.Path tempDir; + + @Test + void testCountsOnlyCommittedDataFiles() 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); + + assertThat(measured.fileCount()).isEqualTo(1); + assertThat(measured.fileSizeInBytes()).isEqualTo(100); + assertThat(measured.lastFileCreationTime()).isPositive(); + } + + @Test + void testStagingTreesAreNotMeasured() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + write(fileIO, tablePath, PARTITION_DIR + "/data-0.csv", 100); + // Every one of these carries an ordinary data file name; only the directory above says + // the file was never committed. + write( + fileIO, + tablePath, + PARTITION_DIR + "/_temporary/0/_temporary/attempt_0/part-0.csv", + 7); + write( + fileIO, + tablePath, + PARTITION_DIR + "/__magic_job-1/tasks/attempt_1/__base/part-1.csv", + 11); + write(fileIO, tablePath, PARTITION_DIR + "/.hive-staging_1/-ext-10000/part-2.csv", 13); + + PartitionStatistics measured = measure(fileIO, tablePath); + + assertThat(measured.fileCount()).isEqualTo(1); + assertThat(measured.fileSizeInBytes()).isEqualTo(100); + } + + @Test + void testHiddenFilesBesideTheDataAreNotMeasured() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + write(fileIO, tablePath, PARTITION_DIR + "/data-0.csv", 100); + write(fileIO, tablePath, PARTITION_DIR + "/_SUCCESS", 0); + write(fileIO, tablePath, PARTITION_DIR + "/.data-0.csv.crc", 8); + + PartitionStatistics measured = measure(fileIO, tablePath); + + assertThat(measured.fileCount()).isEqualTo(1); + assertThat(measured.fileSizeInBytes()).isEqualTo(100); + } + + @Test + void testAMissingDirectoryHasNoFilesAndNoCreationTime() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + + PartitionStatistics measured = measure(fileIO, tablePath); + + assertThat(measured.fileSizeInBytes()).isZero(); + assertThat(measured.fileCount()).isZero(); + // A listing never opens a file, so it has not learned that this partition holds no rows. + assertThat(PartitionStatistics.isKnown(measured.recordCount())).isFalse(); + // There is no last file, so dating one would be an invention. + assertThat(PartitionStatistics.isKnown(measured.lastFileCreationTime())).isFalse(); + } + + @Test + void testADirectoryHoldingOnlyStagedFilesMeasuresAsEmpty() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + write( + fileIO, + tablePath, + PARTITION_DIR + "/_temporary/0/_temporary/attempt_0/part-0.csv", + 7); + + PartitionStatistics measured = measure(fileIO, tablePath); + + assertThat(measured.fileCount()).isZero(); + assertThat(measured.fileSizeInBytes()).isZero(); + assertThat(PartitionStatistics.isKnown(measured.recordCount())).isFalse(); + } + + @Test + void testTheValueOnlyLayoutIsMeasuredWhereItsFilesActuallyAre() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + // The same partition, laid out with values only. A measurement that assumed key=value + // would look at a directory that does not exist and call the partition empty. + write(fileIO, tablePath, "2025/10/data-0.csv", 64); + + Map options = new HashMap<>(); + options.put(CoreOptions.FILE_FORMAT.key(), "csv"); + options.put(CoreOptions.FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH.key(), "true"); + PartitionStatistics measured = + new FormatTablePartitionStatsCollector(table(fileIO, tablePath, options), 1) + .collect(Collections.singletonList(spec("2025", "10"))) + .get(0); + + assertThat(measured.fileCount()).isEqualTo(1); + assertThat(measured.fileSizeInBytes()).isEqualTo(64); + } + + @Test + void testAValueThatHasToBeEscapedIsMeasuredWhereItsFilesActuallyAre() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + // The spec carries the raw value, the directory carries the escaped one: a measurement + // that joined the raw value straight into the path would miss the files entirely. + write(fileIO, tablePath, "year=2025/month=a%3Ab/data-0.csv", 32); + + PartitionStatistics measured = + new FormatTablePartitionStatsCollector(table(fileIO, tablePath), 1) + .collect(Collections.singletonList(spec("2025", "a:b"))) + .get(0); + + assertThat(measured.fileCount()).isEqualTo(1); + assertThat(measured.fileSizeInBytes()).isEqualTo(32); + } + + @Test + void testTheResultIsAlignedToTheGivenPartitions() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + write(fileIO, tablePath, "year=2025/month=10/data-0.csv", 100); + write(fileIO, tablePath, "year=2025/month=12/data-0.csv", 200); + List> partitions = + Arrays.asList(spec("2025", "12"), spec("2025", "11"), spec("2025", "10")); + + List measured = + new FormatTablePartitionStatsCollector(table(fileIO, tablePath), 1) + .collect(partitions); + + assertThat(measured).hasSize(3); + assertThat(measured.get(0).spec()).isEqualTo(spec("2025", "12")); + assertThat(measured.get(0).fileSizeInBytes()).isEqualTo(200); + assertThat(measured.get(1).spec()).isEqualTo(spec("2025", "11")); + assertThat(measured.get(1).fileCount()).isZero(); + assertThat(measured.get(2).spec()).isEqualTo(spec("2025", "10")); + assertThat(measured.get(2).fileSizeInBytes()).isEqualTo(100); + } + + @Test + void testParallelCollectionMeasuresTheSameThingAsSerialCollection() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + List> partitions = new ArrayList<>(); + for (int month = 1; month <= 6; month++) { + String dir = String.format("year=2025/month=%02d", month); + write(fileIO, tablePath, dir + "/data-0.csv", month * 10); + write(fileIO, tablePath, dir + "/_temporary/0/attempt_0/part-0.csv", 5); + partitions.add(spec("2025", String.format("%02d", month))); + } + + List serial = + new FormatTablePartitionStatsCollector(table(fileIO, tablePath), 1) + .collect(partitions); + List parallel = + new FormatTablePartitionStatsCollector(table(fileIO, tablePath), 4) + .collect(partitions); + + for (int i = 0; i < partitions.size(); i++) { + assertThat(parallel.get(i).spec()).isEqualTo(serial.get(i).spec()); + assertThat(parallel.get(i).fileCount()).isEqualTo(serial.get(i).fileCount()); + assertThat(parallel.get(i).fileSizeInBytes()) + .isEqualTo(serial.get(i).fileSizeInBytes()); + } + assertThat(serial.get(0).fileSizeInBytes()).isEqualTo(10); + assertThat(serial.get(5).fileSizeInBytes()).isEqualTo(60); + } + + @Test + void testAListingFailureAbortsTheWholeCollection() throws Exception { + IOException listFailure = new IOException("injected partition LIST failure"); + LocalFileIO fileIO = + new LocalFileIO() { + @Override + public FileStatus[] listStatus(Path path) throws IOException { + if ("month=11".equals(path.getName())) { + throw listFailure; + } + return super.listStatus(path); + } + }; + Path tablePath = new Path(tempDir.toUri()); + write(fileIO, tablePath, "year=2025/month=10/data-0.csv", 100); + write(fileIO, tablePath, "year=2025/month=11/data-0.csv", 200); + List> partitions = + Arrays.asList(spec("2025", "10"), spec("2025", "11")); + + // A truncated listing cannot be told apart from a partition that lost files, so nothing at + // all is reported: returning what was measured would write an exact zero over a partition + // that was never read. Both the serial and the parallel path have to abort. + for (int parallelism : new int[] {1, 2}) { + assertThatThrownBy( + () -> + new FormatTablePartitionStatsCollector( + table(fileIO, tablePath), parallelism) + .collect(partitions)) + .isInstanceOf(UncheckedIOException.class) + .hasMessageContaining("month=11") + .hasCause(listFailure); + } + } + + @Test + void testASpecMissingAPartitionKeyIsRejected() { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + + // Without the guard the directory name would carry the literal "null" and the measurement + // would describe a path no reader ever visits. + assertThatThrownBy( + () -> + new FormatTablePartitionStatsCollector(table(fileIO, tablePath), 1) + .collect( + Collections.singletonList( + Collections.singletonMap("year", "2025")))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("month"); + } + + private PartitionStatistics measure(FileIO fileIO, Path tablePath) { + return new FormatTablePartitionStatsCollector(table(fileIO, tablePath), 1) + .collect(Collections.singletonList(spec("2025", "10"))) + .get(0); + } + + private FormatTable table(FileIO fileIO, Path tablePath) { + return table(fileIO, tablePath, FormatTable.Format.CSV, "csv"); + } + + private FormatTable table(FileIO fileIO, Path tablePath, Map options) { + return table(fileIO, tablePath, FormatTable.Format.CSV, options); + } + + private FormatTable table( + FileIO fileIO, Path tablePath, FormatTable.Format format, String fileFormat) { + return table( + fileIO, + tablePath, + format, + Collections.singletonMap(CoreOptions.FILE_FORMAT.key(), fileFormat)); + } + + private FormatTable table( + FileIO fileIO, Path tablePath, FormatTable.Format format, Map options) { + RowType rowType = + RowType.builder() + .field("year", DataTypes.STRING()) + .field("month", DataTypes.STRING()) + .field("id", DataTypes.INT()) + .build(); + return FormatTable.builder() + .fileIO(fileIO) + .identifier(TABLE) + .rowType(rowType) + .partitionKeys(Arrays.asList("year", "month")) + .location(tablePath.toString()) + .format(format) + .options(options) + .build(); + } + + private static void write(FileIO fileIO, Path tablePath, String relativePath, int bytes) + throws Exception { + Path path = new Path(tablePath, relativePath); + fileIO.mkdirs(path.getParent()); + try (PositionOutputStream out = fileIO.newOutputStream(path, false)) { + out.write(new byte[bytes]); + } + } + + private static Map spec(String year, String month) { + LinkedHashMap spec = new LinkedHashMap<>(); + spec.put("year", year); + spec.put("month", month); + return spec; + } +} 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 108fe12dac79..4c0ec743ff4f 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 @@ -155,6 +155,23 @@ public class SparkConnectorOptions { .withDescription( "Whether to allow full scan when reading a partitioned table."); + public static final ConfigOption FORMAT_TABLE_REPAIR_COLLECT_STATISTICS = + key("format-table.repair.collect-statistics") + .booleanType() + .defaultValue(false) + .withDescription( + "Whether MSCK REPAIR TABLE on a Format Table also measures the partitions it " + + "finds. Off by default: measuring lists the files inside every partition, " + + "not only the partition directories."); + + public static final ConfigOption FORMAT_TABLE_STATISTICS_PARALLELISM = + key("format-table.statistics.parallelism") + .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."); + public static final ConfigOption SOURCE_SPLIT_TARGET_SIZE_WITH_COLUMN_PRUNING = key("source.split.target-size-with-column-pruning") .booleanType() diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/format/FormatTablePartitionRepair.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/format/FormatTablePartitionRepair.java index d77968f248dd..d213f4d86b29 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/format/FormatTablePartitionRepair.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/format/FormatTablePartitionRepair.java @@ -23,10 +23,13 @@ import org.apache.paimon.partition.Partition; import org.apache.paimon.table.FormatTable; import org.apache.paimon.table.format.FormatTablePartitionManager; +import org.apache.paimon.table.format.FormatTablePartitionStatsCollector; import org.apache.paimon.utils.Pair; import org.apache.paimon.utils.PartitionPathUtils; import org.apache.paimon.utils.Preconditions; +import javax.annotation.Nullable; + import java.util.ArrayList; import java.util.Collections; import java.util.Comparator; @@ -52,6 +55,23 @@ private FormatTablePartitionRepair() {} */ public static int repair( PaimonFormatTable sparkTable, boolean addPartitions, boolean dropPartitions) { + return repair(sparkTable, addPartitions, dropPartitions, null); + } + + /** + * Repair the partition metadata of a Format Table with catalog-managed partitions, optionally + * measuring the partitions it finds and reporting their statistics. + * + *

When measuring, every partition found on the filesystem is measured, not only the newly + * registered ones: a repair is exactly the moment the catalog numbers are known to be behind. + * + * @param statsCollector measures the partitions, or null to only reconcile the registration + */ + public static int repair( + PaimonFormatTable sparkTable, + boolean addPartitions, + boolean dropPartitions, + @Nullable FormatTablePartitionStatsCollector statsCollector) { Preconditions.checkArgument( addPartitions || dropPartitions, "MSCK REPAIR TABLE must enable ADD and/or DROP partitions"); @@ -67,15 +87,14 @@ public static int repair( listFilesystemPartitionSpecs(formatTable), formatTable.partitionKeys(), addPartitions, - dropPartitions); + dropPartitions, + statsCollector); } private static List> listFilesystemPartitionSpecs(FormatTable formatTable) { - // Discover partitions from the raw directory names rather than through the table scan: - // the scan casts each value to its column type and back (e.g. month=01 -> 1), producing - // specs that can no longer round-trip to the real directory. The write path registers the - // raw directory value, so a repair must diff against the same raw values to avoid - // spuriously adding/dropping partition metadata. + // Raw directory names rather than the table scan: the scan casts each value to its column + // type and back (month=01 -> 1), producing specs that no longer name the real directory, + // while the write path registers the raw value. boolean onlyValueInPath = new CoreOptions(formatTable.options()).formatTablePartitionOnlyValueInPath(); List, Path>> found = @@ -96,11 +115,9 @@ private static List> listFilesystemPartitionSpecs(FormatTabl /** * Diff the filesystem partition set against the catalog registration set and apply the * requested actions. ADD registers "directory exists but unregistered"; DROP is metadata-only - * cleanup of "registered but directory missing". Scan-completeness guard: the filesystem - * listing that feeds {@code filesystemPartitions} ({@link - * PartitionPathUtils#searchPartSpecAndPaths}) fails on any mid-scan LIST error instead of - * returning a truncated set, so a DROP diff can only be produced from a complete listing and a - * transient failure never deregisters partitions that still exist. + * cleanup of "registered but directory missing". {@link + * PartitionPathUtils#searchPartSpecAndPaths} fails on a mid-scan LIST error rather than + * returning a truncated set, so a transient failure never deregisters partitions that exist. */ static int apply( FormatTablePartitionManager partitionManager, @@ -108,6 +125,22 @@ static int apply( List partitionKeys, boolean addPartitions, boolean dropPartitions) { + return apply( + partitionManager, + filesystemPartitions, + partitionKeys, + addPartitions, + dropPartitions, + null); + } + + static int apply( + FormatTablePartitionManager partitionManager, + List> filesystemPartitions, + List partitionKeys, + boolean addPartitions, + boolean dropPartitions, + @Nullable FormatTablePartitionStatsCollector statsCollector) { Set> registeredPartitions = new HashSet<>(); for (Partition partition : partitionManager.listPartitions(Collections.emptyMap(), null)) { @@ -135,11 +168,25 @@ static int apply( sortByCanonicalPath(dropDiff, partitionKeys); } - // A first repair of a pre-existing table can discover far more partitions than any regular - // write. Splitting such a diff into per-request batches is the partition catalog's job; - // registration is an idempotent upsert and unregistration ignores missing partitions, so a + // A first repair can discover far more partitions than any regular write. Splitting the + // diff into requests is the partition catalog's job; both halves are idempotent, so a // mid-way failure leaves a state a rerun converges from. - if (!addDiff.isEmpty()) { + if (statsCollector != null) { + // Every partition that ends up registered with a directory behind it, not only the + // newly added ones: numbers for partitions written outside Paimon are what a repair + // exists to correct. Without ADD it stays inside the already registered set. + List> measured = new ArrayList<>(); + for (Map partition : filesystemPartitions) { + if (addPartitions || registeredPartitions.contains(partition)) { + measured.add(partition); + } + } + sortByCanonicalPath(measured, partitionKeys); + if (!measured.isEmpty()) { + partitionManager.createPartitions( + measured, true, statsCollector.collect(measured), true); + } + } else if (!addDiff.isEmpty()) { partitionManager.createPartitions(addDiff, true); } if (!dropDiff.isEmpty()) { diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonFormatTablePartitionDdlExec.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonFormatTablePartitionDdlExec.scala index 45c925b4469d..84cd80d12d27 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonFormatTablePartitionDdlExec.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonFormatTablePartitionDdlExec.scala @@ -20,6 +20,8 @@ package org.apache.paimon.spark.execution import org.apache.paimon.CoreOptions import org.apache.paimon.spark.format.{FormatTablePartitionRepair, PaimonFormatTable} +import org.apache.paimon.spark.util.OptionUtils +import org.apache.paimon.table.format.FormatTablePartitionStatsCollector import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.catalyst.analysis.{NoSuchPartitionsException, ResolvedPartitionSpec} @@ -189,8 +191,18 @@ case class PaimonRepairFormatTablePartitionsExec( extends LeafV2CommandExec { override protected def run(): Seq[InternalRow] = { + // A repair stops at the numbers a directory listing already gives; a record count needs the + // file footers, which no listing opens. + val statsCollector = + if (OptionUtils.formatTableRepairCollectStatistics()) { + new FormatTablePartitionStatsCollector( + table.table, + OptionUtils.formatTableStatisticsParallelism()) + } else { + null + } PaimonFormatTablePartitionDdlExec.refreshingCache(refreshCache) { - FormatTablePartitionRepair.repair(table, addPartitions, dropPartitions) + FormatTablePartitionRepair.repair(table, addPartitions, dropPartitions, statsCollector) } Seq.empty } diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/OptionUtils.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/OptionUtils.scala index 10d403248bd3..1649a57eadb6 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/OptionUtils.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/OptionUtils.scala @@ -146,6 +146,14 @@ object OptionUtils extends SQLConfHelper with Logging { getOptionString(SparkConnectorOptions.SOURCE_SPLIT_TARGET_SIZE_WITH_COLUMN_PRUNING).toBoolean } + def formatTableRepairCollectStatistics(): Boolean = { + getOptionString(SparkConnectorOptions.FORMAT_TABLE_REPAIR_COLLECT_STATISTICS).toBoolean + } + + def formatTableStatisticsParallelism(): Int = { + getOptionString(SparkConnectorOptions.FORMAT_TABLE_STATISTICS_PARALLELISM).toInt + } + private def mergeSQLConf(extraOptions: JMap[String, String]): JMap[String, String] = { val mergedOptions = new JHashMap[String, String]( conf.getAllConfs diff --git a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/format/FormatTablePartitionRepairTest.java b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/format/FormatTablePartitionRepairTest.java index 033285c46e93..1ee208f2b9ca 100644 --- a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/format/FormatTablePartitionRepairTest.java +++ b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/format/FormatTablePartitionRepairTest.java @@ -29,6 +29,7 @@ import org.apache.paimon.predicate.Predicate; import org.apache.paimon.table.FormatTable; import org.apache.paimon.table.format.FormatTablePartitionManager; +import org.apache.paimon.table.format.FormatTablePartitionStatsCollector; import org.apache.paimon.types.DataTypes; import org.apache.paimon.types.RowType; @@ -38,6 +39,7 @@ import javax.annotation.Nullable; import java.io.IOException; +import java.io.UncheckedIOException; import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.util.ArrayList; @@ -368,6 +370,137 @@ public FileStatus[] listStatus(Path path) throws IOException { assertThat(catalog.droppedPartitions).isEmpty(); } + @Test + void repairMeasuresEveryPartitionOnDiskAndReplacesTheirStatistics() throws Exception { + java.nio.file.Path known = Files.createDirectories(tempDir.resolve("dt=20260701")); + Files.write(known.resolve("data.csv"), Arrays.asList("1", "2"), StandardCharsets.UTF_8); + java.nio.file.Path fresh = Files.createDirectories(tempDir.resolve("dt=20260702")); + Files.write( + fresh.resolve("data.csv"), Collections.singletonList("3"), StandardCharsets.UTF_8); + + RecordingPartitionManager catalog = new RecordingPartitionManager(); + catalog.register(Collections.singletonList(spec("dt", "20260701"))); + FormatTable table = formatTable(tempDir.toUri().toString(), catalog); + PaimonFormatTable sparkTable = new PaimonFormatTable(table); + + int applied = + FormatTablePartitionRepair.repair( + sparkTable, true, false, new FormatTablePartitionStatsCollector(table, 1)); + + // Only one partition was missing from the registration, but a repair that measures corrects + // the numbers of the already-registered one too — being behind is why it is running. + assertThat(applied).isEqualTo(1); + assertThat(catalog.createdPartitions) + .containsExactly(Arrays.asList(spec("dt", "20260701"), spec("dt", "20260702"))); + assertThat(catalog.replaceFlags).containsExactly(true); + // One measurement per spec, in the same order: the catalog reads the two lists side by + // side, so a short or reordered statistics list would describe the wrong partitions. + List reported = catalog.reportedStatistics.get(0); + assertThat(reported).hasSize(2); + assertThat(reported.get(0).spec()).isEqualTo(spec("dt", "20260701")); + assertThat(reported.get(0).fileCount()).isEqualTo(1); + assertThat(reported.get(0).fileSizeInBytes()).isPositive(); + assertThat(reported.get(0).lastFileCreationTime()).isPositive(); + // CSV carries no footer, so the row count is unknown rather than a number nobody measured. + assertThat(PartitionStatistics.isKnown(reported.get(0).recordCount())).isFalse(); + assertThat(reported.get(1).spec()).isEqualTo(spec("dt", "20260702")); + assertThat(reported.get(1).fileCount()).isEqualTo(1); + assertThat(reported.get(1).fileSizeInBytes()).isPositive(); + assertThat(catalog.droppedPartitions).isEmpty(); + } + + @Test + void repairWritesNothingWhenMeasuringAPartitionFailsToList() throws Exception { + Files.write( + Files.createDirectories(tempDir.resolve("dt=20260701")).resolve("data.csv"), + Collections.singletonList("1"), + StandardCharsets.UTF_8); + Files.createDirectories(tempDir.resolve("dt=20260702")); + + IOException listFailure = new IOException("injected partition measurement LIST failure"); + FileIO fileIO = + new LocalFileIO() { + @Override + public FileStatus[] listStatus(Path path) throws IOException { + if ("dt=20260702".equals(path.getName())) { + throw listFailure; + } + return super.listStatus(path); + } + }; + + RecordingPartitionManager catalog = new RecordingPartitionManager(); + FormatTable table = formatTable(fileIO, tempDir.toUri().toString(), false, catalog); + PaimonFormatTable sparkTable = new PaimonFormatTable(table); + + assertThatThrownBy( + () -> + FormatTablePartitionRepair.repair( + sparkTable, + true, + false, + new FormatTablePartitionStatsCollector(table, 1))) + .isInstanceOf(UncheckedIOException.class) + .hasCause(listFailure); + // The partition that did list measured fine, but half a measurement written as if it were + // the whole one is the corruption the abort exists to prevent: nothing reaches the catalog, + // and the registration the repair would have added is not applied either. + assertThat(catalog.createdPartitions).isEmpty(); + assertThat(catalog.reportedStatistics).isEmpty(); + assertThat(catalog.droppedPartitions).isEmpty(); + } + + @Test + void repairWithoutAddNeverRegistersAPartitionJustToMeasureIt() throws Exception { + java.nio.file.Path registeredDirectory = + Files.createDirectories(tempDir.resolve("dt=20260701")); + Files.write( + registeredDirectory.resolve("data.csv"), + Collections.singletonList("1"), + StandardCharsets.UTF_8); + java.nio.file.Path unregisteredDirectory = + Files.createDirectories(tempDir.resolve("dt=20260702")); + Files.write( + unregisteredDirectory.resolve("data.csv"), + Collections.singletonList("2"), + StandardCharsets.UTF_8); + + RecordingPartitionManager catalog = new RecordingPartitionManager(); + catalog.register(Collections.singletonList(spec("dt", "20260701"))); + FormatTable table = formatTable(tempDir.toUri().toString(), catalog); + PaimonFormatTable sparkTable = new PaimonFormatTable(table); + + FormatTablePartitionRepair.repair( + sparkTable, false, true, new FormatTablePartitionStatsCollector(table, 1)); + + // MSCK DROP PARTITIONS asked for no registrations; measuring must not smuggle one in. + assertThat(catalog.createdPartitions) + .containsExactly(Collections.singletonList(spec("dt", "20260701"))); + } + + @Test + void repairWithoutMeasuringKeepsTheSpecOnlyRegistration() throws Exception { + java.nio.file.Path partitionDirectory = + Files.createDirectories(tempDir.resolve("dt=20260701")); + Files.write( + partitionDirectory.resolve("data.csv"), + Collections.singletonList("1"), + StandardCharsets.UTF_8); + + RecordingPartitionManager catalog = new RecordingPartitionManager(); + PaimonFormatTable sparkTable = + new PaimonFormatTable(formatTable(tempDir.toUri().toString(), catalog)); + + FormatTablePartitionRepair.repair(sparkTable, true, false); + + assertThat(catalog.createdPartitions) + .containsExactly(Collections.singletonList(spec("dt", "20260701"))); + // Registering without measuring is one call that carries no statistics, not the absence of + // a call: the repair still has to register what it found. + assertThat(catalog.reportedStatistics).hasSize(1).containsOnlyNulls(); + assertThat(catalog.replaceFlags).containsExactly(false); + } + private static Map spec(String key, String value) { Map spec = new LinkedHashMap<>(); spec.put(key, value); @@ -380,13 +513,21 @@ private static FormatTable formatTable(String location, FormatTablePartitionMana private static FormatTable formatTable( String location, boolean onlyValueInPath, FormatTablePartitionManager catalog) { + return formatTable(LocalFileIO.create(), location, onlyValueInPath, catalog); + } + + private static FormatTable formatTable( + FileIO fileIO, + String location, + boolean onlyValueInPath, + FormatTablePartitionManager catalog) { RowType rowType = RowType.builder() .field("id", DataTypes.INT()) .field("dt", DataTypes.STRING()) .build(); return build( - LocalFileIO.create(), + fileIO, location, rowType, Collections.singletonList("dt"), @@ -439,6 +580,8 @@ private static class RecordingPartitionManager implements FormatTablePartitionMa private final List>> createdPartitions = new ArrayList<>(); private final List createIgnoreFlags = new ArrayList<>(); private final List>> droppedPartitions = new ArrayList<>(); + private final List> reportedStatistics = new ArrayList<>(); + private final List replaceFlags = new ArrayList<>(); private void register(List> partitions) { registered.addAll(partitions); @@ -452,6 +595,8 @@ public void createPartitions( boolean replaceStatistics) { createdPartitions.add(new ArrayList<>(partitions)); createIgnoreFlags.add(ignoreIfExists); + reportedStatistics.add(statistics == null ? null : new ArrayList<>(statistics)); + replaceFlags.add(replaceStatistics); } @Override diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionMsckRepairTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionMsckRepairTest.scala index 586099b73b38..bee92ed56226 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionMsckRepairTest.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionMsckRepairTest.scala @@ -22,7 +22,7 @@ import org.apache.paimon.catalog.Identifier import org.apache.paimon.fs.Path import org.apache.paimon.partition.{Partition, PartitionStatistics} import org.apache.paimon.predicate.Predicate -import org.apache.paimon.spark.{PaimonSparkTestWithRestCatalogBase, SparkCatalog} +import org.apache.paimon.spark.{PaimonSparkTestWithRestCatalogBase, SparkCatalog, SparkConnectorOptions} import org.apache.paimon.spark.execution.PaimonRepairFormatTablePartitionsExec import org.apache.paimon.spark.format.PaimonFormatTable import org.apache.paimon.table.FormatTable @@ -415,6 +415,46 @@ class CatalogManagedPartitionMsckRepairTest extends PaimonSparkTestWithRestCatal } } + test("MSCK leaves statistics unknown until it is asked to measure") { + val tableName = "msck_statistics" + val partition = "20260721" + + withTable(tableName) { + createFormatTableWithCatalogManagedPartitions(tableName) + writeCsvPartition(tableName, partition, 21, "measured") + + executeCatalogManagedRepair(s"MSCK REPAIR TABLE paimon.$dbName0.$tableName") + // Registering a partition measures nothing about it, so every statistic stays unknown — + // an exact zero here would be a number nobody took. + val registered = statisticsOf(tableName, partition) + assert(!PartitionStatistics.isKnown(registered.fileCount()), registered.toString) + assert(!PartitionStatistics.isKnown(registered.fileSizeInBytes()), registered.toString) + assert(!PartitionStatistics.isKnown(registered.recordCount()), registered.toString) + + val collectStatistics = + s"spark.paimon.${SparkConnectorOptions.FORMAT_TABLE_REPAIR_COLLECT_STATISTICS.key()}" + withSQLConf(collectStatistics -> "true") { + executeCatalogManagedRepair(s"MSCK REPAIR TABLE paimon.$dbName0.$tableName") + } + + val measured = statisticsOf(tableName, partition) + assert(measured.fileCount() == 1L, measured.toString) + assert(measured.fileSizeInBytes() > 0L, measured.toString) + assert(measured.lastFileCreationTime() > 0L, measured.toString) + // A repair only lists; CSV carries no footer, so the row count is still nobody's measurement. + assert(!PartitionStatistics.isKnown(measured.recordCount()), measured.toString) + // Measuring is not a way to change which partitions exist. + assertPartitionState(tableName, Set(partition)) + } + } + + private def statisticsOf(tableName: String, partition: String): Partition = + paimonCatalog + .listPartitions(tableIdentifier(tableName)) + .asScala + .find(_.spec().get("dt") == partition) + .getOrElse(fail(s"partition dt=$partition of $tableName is not registered")) + private def createFormatTableWithCatalogManagedPartitions(tableName: String): Unit = sql(s"""CREATE TABLE $tableName (id INT, payload STRING, dt STRING) |USING CSV @@ -648,7 +688,7 @@ private[sql] class FaultInjectingFormatTablePartitionManager(delegate: FormatTab ignoreIfExists: Boolean, statistics: JList[PartitionStatistics], replaceStatistics: Boolean): Unit = { - delegate.createPartitions(partitions, ignoreIfExists) + delegate.createPartitions(partitions, ignoreIfExists, statistics, replaceStatistics) MsckFaultInjection.createCalls += 1 } From 77ae6ee81f02ef8b6d96058d7a8593477306f630 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 2/2] [spark] Support ANALYZE TABLE on catalog-managed format tables MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Spark rejects ANALYZE TABLE for every v2 table, so a user reaching for the standard command on a format table gets told it is unsupported. It is the command that exists to recompute statistics, and after this it does. What it means here is narrower than for a Paimon table, and the difference is worth stating. A format table has no snapshot to carry table-level statistics and no column statistics to compute; analyzing it measures its partitions. So both modifiers carry weight: NOSCAN stops at the listing, and a full ANALYZE also reads file footers for exact row counts — which no listing can give and no commit can give for data it did not write. Reading those footers is what this commit adds to the collector MSCK REPAIR already uses: a row counter built from the table's format, a per-file read of it, and the rule that one unreadable file leaves the whole partition's row count unknown rather than short. A repair never asks for it, so until now nothing did, and the flag that asks arrives here with the command that pays for it. PARTITION(...) must name a leading run of the partition keys, because that is the shape the catalog can select on. Anything else is rejected rather than quietly widened to more partitions than the user asked about; naming a partition that does not exist is rejected too, rather than reporting success for having measured nothing. Following Spark, a partition column named without a value means every value of it, and the column names are resolved case-insensitively like every other command's. A format table that discovers its partitions from the filesystem has no catalog to write to and keeps the upstream rejection. Measuring replaces rather than accumulates, so running it twice is running it once. Row counts stay unknown for formats without a footer, and a footer that cannot be read makes the whole partition's row count unknown rather than short. Tests: CatalogManagedPartitionAnalyzeTest covers the command end to end — a full spec measuring only its own partition and a leading prefix measuring every partition under it, a non-leading column and a column that is not a partition column both rejected rather than silently widened, a partition that is not registered failing instead of reporting success for having measured nothing, column names resolved the way the rest of Spark resolves them, NOSCAN keeping a row count that is already known while a full ANALYZE reads the one a NOSCAN cannot, a second run reporting the same measurement rather than accumulating, files a committer left staged not being counted, the registered set never changing, and a filesystem-discovering format table keeping the upstream rejection. --- docs/docs/spark/auxiliary.md | 4 + docs/docs/spark/sql-ddl.md | 22 +- .../FormatTablePartitionStatsCollector.java | 111 ++++- ...ormatTablePartitionStatsCollectorTest.java | 116 ++++- .../catalyst/analysis/PaimonAnalysis.scala | 12 +- ...nAnalyzeFormatTablePartitionsCommand.scala | 119 +++++ .../CatalogManagedPartitionAnalyzeTest.scala | 407 ++++++++++++++++++ 7 files changed, 779 insertions(+), 12 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..1ccf9c004b5e 100644 --- a/docs/docs/spark/auxiliary.md +++ b/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. diff --git a/docs/docs/spark/sql-ddl.md b/docs/docs/spark/sql-ddl.md index 4ac28fd6521e..b0901f74ad8f 100644 --- a/docs/docs/spark/sql-ddl.md +++ b/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 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..95f1e2519433 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,23 @@ 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; @@ -38,6 +44,7 @@ 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; @@ -61,15 +68,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 +108,19 @@ public List collect(List> 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 statistics = new ArrayList<>(partitions.size()); for (Map partition : partitions) { - statistics.add(measure(partition)); + statistics.add(measure(partition, rowCounter)); } return statistics; } @@ -95,7 +130,7 @@ public List collect(List> partitions) { try { List> futures = new ArrayList<>(partitions.size()); for (Map partition : partitions) { - futures.add(executor.submit(() -> measure(partition))); + futures.add(executor.submit(() -> measure(partition, rowCounter))); } List statistics = new ArrayList<>(partitions.size()); for (Future future : futures) { @@ -116,7 +151,8 @@ public List collect(List> partitions) { } } - private PartitionStatistics measure(Map partition) { + private PartitionStatistics measure( + Map partition, @Nullable SimpleStatsExtractor rowCounter) { FileIO fileIO = table.fileIO(); Path partitionPath = partitionPath(partition); List files; @@ -126,7 +162,7 @@ private PartitionStatistics measure(Map partition) { files = FormatTableScan.listDataFiles(fileIO, partitionPath); } catch (FileNotFoundException e) { // A registered partition whose directory is gone reads as empty. - return empty(partition); + return empty(partition, rowCounter); } catch (IOException e) { throw new UncheckedIOException( String.format( @@ -140,29 +176,86 @@ private PartitionStatistics measure(Map partition) { 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); + return empty(partition, rowCounter); } return new PartitionStatistics( partition, - // A listing never opens a file, so the rows a partition holds stay unknown. - PartitionStatistics.UNKNOWN, + 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 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 with nothing in it: the file numbers are an exact zero. */ - private static PartitionStatistics empty(Map partition) { + private static PartitionStatistics empty( + Map partition, @Nullable SimpleStatsExtractor rowCounter) { return new PartitionStatistics( partition, - PartitionStatistics.UNKNOWN, + // 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, 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..dd6d3f4a0b25 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,11 @@ 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.format.SupportsDirectWrite; import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.FileStatus; import org.apache.paimon.fs.Path; @@ -287,7 +292,11 @@ void testASpecMissingAPartitionKeyIsRejected() { } 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 +305,111 @@ private FormatTable table(FileIO fileIO, Path tablePath) { return table(fileIO, tablePath, FormatTable.Format.CSV, "csv"); } + @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 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 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()); + if (factory instanceof SupportsDirectWrite) { + FormatWriter writer = ((SupportsDirectWrite) factory).create(fileIO, path, "zstd"); + for (int i = 0; i < rows; i++) { + writer.addElement(GenericRow.of(i)); + } + writer.close(); + } else { + 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/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..18268ce9acc7 --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeFormatTablePartitionsCommand.scala @@ -0,0 +1,119 @@ +/* + * 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.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 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 the result replaces what the catalog holds, so this + * is how a table drifts back into agreement after 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. Formats that carry no footer (CSV, TEXT, + * JSON) keep an unknown row count either way rather than a guessed 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 a leading subset of them. + */ +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().asInstanceOf[JMap[String, String]]) + .toList + + if (partitions.isEmpty && prefix.nonEmpty) { + throw new IllegalArgumentException( + s"Partition ${prefix.map { case (k, v) => s"$k=$v" }.mkString("(", ", ", ")")} of table " + + s"${v2Table.name()} does not exist, so there is nothing to measure.") + } + + 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: column names resolve + * under the session's case sensitivity, 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 resolver = sparkSession.sessionState.conf.resolver + val partitionKeys = v2Table.table.partitionKeys().asScala.toSeq + val normalized = partitionSpec.map { + case (key, value) => + val resolved = partitionKeys + .find(partitionKey => resolver(partitionKey, key)) + .getOrElse( + throw new IllegalArgumentException( + s"$key is not a partition column of ${v2Table.name()}, whose partition columns are " + + partitionKeys.mkString("[", ", ", "]"))) + resolved -> value + } + 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 " + + partitionSpec.keys.mkString("[", ", ", "]")) + } + // Kept in partition-key order, so a message built from it reads in that order too. + ListMap(prefix.map { case (key, value) => key -> value.get }: _*) + } +} 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..1e944e376b20 --- /dev/null +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionAnalyzeTest.scala @@ -0,0 +1,407 @@ +/* + * 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 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() + } + assert(causeMessages(error).contains("20991231"), causeMessages(error)) + assert( + causeMessages(error).contains("does not exist, so there is nothing to measure"), + causeMessages(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 is not a partition column"), + 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) + } + } + + 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'), + | (3, 'c', '20260102', '00') + |""".stripMargin) + // The write registered both partitions and, with reporting off, measured neither. + assert(!PartitionStatistics.isKnown(statisticsOf(tableName, "20260101", "00").recordCount())) + + sql(s"ANALYZE TABLE ${qualified(tableName)} PARTITION (dt = '20260101') COMPUTE STATISTICS") + .collect() + + // No NOSCAN, so the footers were read and the row count is exact. + val scanned = statisticsOf(tableName, "20260101", "00") + assert(scanned.recordCount() == 2L, scanned.toString) + assert(scanned.fileCount() >= 1L, scanned.toString) + + sql( + s"ANALYZE TABLE ${qualified(tableName)} PARTITION (dt = '20260102') " + + 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, "20260102", "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 + } + + 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 causeMessages(error: Throwable): String = + Iterator + .iterate(error)(_.getCause) + .takeWhile(_ != null) + .map(e => String.valueOf(e.getMessage)) + .mkString(" | ") +}