From e76a3784a4920df66157d8cd33696d89370e60ed 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] [core][spark] Support collecting format table partition statistics 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 }