From 5212eb81a2c4f16da991b0bcdc9b8f0a5f4033cd 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] 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 needs a file footer, which is what ANALYZE is for. 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. An empty partition reports the row count as unknown unless the format can count rows. The file numbers are an exact zero either way, but a listing that never opens a file has not learned that the partition holds no rows, and a zero it reported would outlive the emptiness: the measurement that follows the files arriving reports the row count as unknown, and an unknown replaces nothing, so the catalog would keep claiming no rows for a partition that has them. Tests: FormatTablePartitionStatsCollectorTest covers the staging trees a committer leaves behind, exact parquet row counts, an unreadable footer, a missing directory, the one-for-one alignment of the result with the given specs, a spec that omits a partition key, and a listing failure aborting the whole collection on both the serial and the parallel path. FormatTablePartitionRepairTest covers measuring every partition on disk rather than only the additions, 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 | 296 ++++++++++++ ...ormatTablePartitionStatsCollectorTest.java | 431 ++++++++++++++++++ .../paimon/spark/SparkConnectorOptions.java | 22 + .../format/FormatTablePartitionRepair.java | 61 ++- .../PaimonFormatTablePartitionDdlExec.scala | 15 +- .../paimon/spark/util/OptionUtils.scala | 8 + .../FormatTablePartitionRepairTest.java | 150 +++++- ...atalogManagedPartitionMsckRepairTest.scala | 44 +- 9 files changed, 1033 insertions(+), 6 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..2eb674ccce34 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 and reports their statistics to the catalog. Off by default, because it changes what a repair costs: the plain repair lists partition directories, while measuring lists the files inside every one of them. + + +
format-table.statistics.parallelism
+ 8 + Integer + How many Format Table partitions MSCK REPAIR TABLE and ANALYZE TABLE measure at once. Listing a partition is one round trip to storage, so the driver spends its time waiting; the ceiling keeps that from turning into a burst of requests against a table with many partitions. This is the driver-side measuring path, separate from 'format-table.scan.list-parallelism', which governs the same per-partition listing on the read path. +
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..4e076685be89 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionStatsCollector.java @@ -0,0 +1,296 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.table.format; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.format.FileFormat; +import org.apache.paimon.format.SimpleStatsExtractor; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.FileStatus; +import org.apache.paimon.fs.Path; +import org.apache.paimon.partition.PartitionStatistics; +import org.apache.paimon.statistics.SimpleColStatsCollector; +import org.apache.paimon.table.FormatTable; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.PartitionPathUtils; +import org.apache.paimon.utils.ThreadPoolUtils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.io.FileNotFoundException; +import java.io.IOException; +import java.io.UncheckedIOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; + +/** + * Measures what the partitions of a Format Table currently hold, by listing their directories. + * + *

File count, byte size and last file creation time come from the listing itself. A listing only + * reports a modification time, which for the immutable data files of a Format Table is when the + * file was created; a file rewritten in place therefore dates its partition by the rewrite. This is + * a different clock from the one a managed table records in {@link + * org.apache.paimon.manifest.PartitionEntry}, which carries the creation time the writer stamped. + * The row count needs the file footer and is only available for the formats that carry one — a CSV, + * TEXT or JSON partition keeps an unknown row count rather than a guessed one. A partition that + * holds nothing, whether its directory is empty or gone, measures as an exact zero on the three + * counts; its last file creation time stays unknown, because there is no last file to date. + * + *

It measures through {@link FormatTableScan#listDataFiles}, the listing the scan itself uses, + * so a measurement counts exactly the files a reader would return. Committer staging trees are + * pruned rather than walked: the files under them carry ordinary data file names, and a measurement + * that counted them would report a partition holding more than any query returns. + * + *

The result is a whole-partition measurement, so it is reported as a replacement rather than an + * increment. This never decides that a partition should exist or stop existing: it only measures + * the ones it is given. + * + *

Listing failures abort the whole collection rather than yield a partial measurement — a + * truncated listing looks exactly like a partition that lost files. + */ +public class FormatTablePartitionStatsCollector { + + private static final Logger LOG = + LoggerFactory.getLogger(FormatTablePartitionStatsCollector.class); + + /** Row counts that need no columns: the file footer alone answers how many rows it holds. */ + private static final RowType NO_COLUMNS = RowType.builder().build(); + + private static final SimpleColStatsCollector.Factory[] NO_COLLECTORS = + new SimpleColStatsCollector.Factory[0]; + + private final FormatTable table; + private final boolean onlyValueInPath; + private final boolean withRecordCount; + private final int parallelism; + + public FormatTablePartitionStatsCollector( + FormatTable table, boolean withRecordCount, int parallelism) { + this.table = table; + this.onlyValueInPath = + new CoreOptions(table.options()).formatTablePartitionOnlyValueInPath(); + this.withRecordCount = withRecordCount; + this.parallelism = Math.max(1, parallelism); + } + + /** + * Measures the given partitions. The result is aligned to {@code partitions} one for one, so a + * caller can send it straight to the catalog alongside the same specs. + */ + public List 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, rowCounter)); + } + 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, rowCounter))); + } + 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, @Nullable SimpleStatsExtractor rowCounter) { + FileIO fileIO = table.fileIO(); + Path partitionPath = partitionPath(partition); + List files; + try { + // The same listing the scan uses, so a measurement counts exactly what a reader would + // return: staging trees are pruned rather than walked, and no separate existence check + // is needed because a missing directory surfaces here as a FileNotFoundException. + files = FormatTableScan.listDataFiles(fileIO, partitionPath); + } catch (FileNotFoundException e) { + // A registered partition whose directory is gone reads as empty, so measuring it as an + // exact zero says the same thing the reader already does. + return empty(partition, rowCounter); + } catch (IOException e) { + throw new UncheckedIOException( + String.format( + "Failed to list partition %s of table %s; no statistics are written " + + "because a partial listing cannot be told apart from a " + + "partition that lost files.", + partitionPath, table.fullName()), + e); + } + + long fileCount = 0; + long fileSizeInBytes = 0; + long lastFileCreationTime = 0; + long recordCount = rowCounter == null ? PartitionStatistics.UNKNOWN : 0L; + for (FileStatus file : files) { + fileCount++; + fileSizeInBytes += file.getLen(); + lastFileCreationTime = Math.max(lastFileCreationTime, file.getModificationTime()); + if (PartitionStatistics.isKnown(recordCount)) { + long rows = rowCount(rowCounter, file); + recordCount = + PartitionStatistics.isKnown(rows) + ? recordCount + rows + : PartitionStatistics.UNKNOWN; + } + } + if (fileCount == 0) { + return empty(partition, rowCounter); + } + return new PartitionStatistics( + partition, + recordCount, + fileSizeInBytes, + fileCount, + lastFileCreationTime, + PartitionStatistics.UNKNOWN_TOTAL_BUCKETS); + } + + /** + * A partition holding nothing. The counts are an exact zero, but there is no last file to date, + * so that one field stays unknown rather than claiming the epoch. + */ + /** + * A partition with nothing in it. The file numbers are an exact zero, but the record count is + * only zero to a measurement that counts records at all: a listing that never opens a file has + * not learned that this partition holds no rows, and a zero it reports would outlive the + * emptiness, since a later measurement of the same kind reports the record count as unknown and + * an unknown replaces nothing. + */ + private static PartitionStatistics empty( + Map partition, @Nullable SimpleStatsExtractor rowCounter) { + return new PartitionStatistics( + partition, + rowCounter == null ? PartitionStatistics.UNKNOWN : 0L, + 0L, + 0L, + PartitionStatistics.UNKNOWN, + PartitionStatistics.UNKNOWN_TOTAL_BUCKETS); + } + + /** + * Rows in one file, or unknown when its footer cannot be read. One unreadable file makes the + * whole partition unknown rather than short: a sum missing a file, reported as exact, is worse + * than no number at all. + */ + private long rowCount(SimpleStatsExtractor rowCounter, FileStatus file) { + try { + return rowCounter + .extractWithFileInfo(table.fileIO(), file.getPath(), file.getLen()) + .getRight() + .getRowCount(); + } catch (Exception e) { + LOG.warn( + "Failed to read the row count of {} in table {}; the row count of its " + + "partition stays unknown.", + file.getPath(), + table.fullName(), + e); + return PartitionStatistics.UNKNOWN; + } + } + + /** + * A footer reader for this table's format, or null when the format carries no row count. It is + * built with no columns on purpose: only the file's row count is wanted, and asking for column + * statistics would both cost more and make the reader depend on the file schema matching the + * table's. + */ + @Nullable + private SimpleStatsExtractor rowCounter() { + try { + CoreOptions options = new CoreOptions(table.options()); + Optional extractor = + FileFormat.fileFormat(options).createStatsExtractor(NO_COLUMNS, NO_COLLECTORS); + return extractor.orElse(null); + } catch (Exception e) { + LOG.warn( + "Failed to create a row counter for format {} of table {}; row counts stay " + + "unknown.", + table.format(), + table.fullName(), + e); + return null; + } + } + + private Path partitionPath(Map partition) { + 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..23f10d198276 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTablePartitionStatsCollectorTest.java @@ -0,0 +1,431 @@ +/* + * 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.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; +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, false); + + 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, false); + + assertThat(measured.fileCount()).isEqualTo(1); + assertThat(measured.fileSizeInBytes()).isEqualTo(100); + } + + @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 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, false); + + assertThat(measured.fileCount()).isEqualTo(1); + assertThat(measured.fileSizeInBytes()).isEqualTo(100); + } + + @Test + void testAMissingDirectoryIsAnExactZeroWithNoCreationTime() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + + PartitionStatistics measured = + new FormatTablePartitionStatsCollector(parquetTable(fileIO, tablePath), true, 1) + .collect(Collections.singletonList(spec("2025", "10"))) + .get(0); + + assertThat(measured.recordCount()).isZero(); + assertThat(measured.fileSizeInBytes()).isZero(); + assertThat(measured.fileCount()).isZero(); + // There is no last file, so dating one would be an invention. + assertThat(PartitionStatistics.isKnown(measured.lastFileCreationTime())).isFalse(); + } + + @Test + void testAnEmptyPartitionKeepsTheRowCountUnknownForAFormatThatCannotCountRows() + throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + + PartitionStatistics empty = measure(fileIO, tablePath, true); + + // The file numbers are an exact zero, but a listing that never opens a file has not + // learned that this partition holds no rows. + assertThat(empty.fileCount()).isZero(); + assertThat(empty.fileSizeInBytes()).isZero(); + assertThat(PartitionStatistics.isKnown(empty.recordCount())).isFalse(); + + // A zero here would outlive the emptiness: the measurement that follows the files arriving + // reports the row count as unknown, and an unknown replaces nothing. + write(fileIO, tablePath, PARTITION_DIR + "/part-0.csv", 3); + PartitionStatistics filled = measure(fileIO, tablePath, true); + assertThat(filled.fileCount()).isEqualTo(1); + assertThat(PartitionStatistics.isKnown(filled.recordCount())).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, true); + + assertThat(measured.fileCount()).isZero(); + assertThat(measured.fileSizeInBytes()).isZero(); + assertThat(PartitionStatistics.isKnown(measured.recordCount())).isFalse(); + } + + @Test + void testCsvKeepsAnUnknownRowCountEvenWhenAsked() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + write(fileIO, tablePath, PARTITION_DIR + "/data-0.csv", 100); + + PartitionStatistics measured = measure(fileIO, tablePath, true); + + assertThat(measured.fileCount()).isEqualTo(1); + // CSV carries no footer: an unknown row count beats a guessed one. + assertThat(PartitionStatistics.isKnown(measured.recordCount())).isFalse(); + } + + @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), false, 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), false, 1) + .collect(partitions); + List parallel = + new FormatTablePartitionStatsCollector(table(fileIO, tablePath), false, 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 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 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), false, 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), false, 1) + .collect( + Collections.singletonList( + Collections.singletonMap("year", "2025")))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("month"); + } + + 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); + } + + private FormatTable table(FileIO fileIO, Path tablePath) { + return table(fileIO, tablePath, FormatTable.Format.CSV, "csv"); + } + + private FormatTable parquetTable(FileIO fileIO, Path tablePath) { + return table(fileIO, tablePath, FormatTable.Format.PARQUET, "parquet"); + } + + private FormatTable table( + FileIO fileIO, Path tablePath, FormatTable.Format format, String fileFormat) { + 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(Collections.singletonMap(CoreOptions.FILE_FORMAT.key(), fileFormat)) + .build(); + } + + 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 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..2d2f4a08ef61 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,28 @@ 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 " + + "and reports their statistics to the catalog. Off by default, because it changes " + + "what a repair costs: the plain repair lists partition directories, while " + + "measuring lists the files inside every one of them."); + + 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 and ANALYZE TABLE measure at once. " + + "Listing a partition is one round trip to storage, so the driver spends its time " + + "waiting; the ceiling keeps that from turning into a burst of requests against a " + + "table with many partitions. This is the driver-side measuring path, separate " + + "from 'format-table.scan.list-parallelism', which governs the same per-partition " + + "listing on the read path."); + 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..9906dcdbf6dc 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,26 @@ 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. + * + *

Measuring is off by default because it changes what a repair costs: the plain diff only + * lists partition directories, while measuring lists the files inside every one of them. When + * it is on, the statistics of every partition found on the filesystem are replaced, not just + * those of 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,7 +90,8 @@ public static int repair( listFilesystemPartitionSpecs(formatTable), formatTable.partitionKeys(), addPartitions, - dropPartitions); + dropPartitions, + statsCollector); } private static List> listFilesystemPartitionSpecs(FormatTable formatTable) { @@ -108,6 +132,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)) { @@ -139,7 +179,24 @@ static int apply( // 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 // mid-way failure leaves a state a rerun converges from. - if (!addDiff.isEmpty()) { + if (statsCollector != null) { + // Measuring covers every partition that ends up registered with a directory behind it, + // not just the newly added ones: the catalog numbers for partitions written outside it + // are exactly what a repair exists to correct. The measurement is of a whole partition, + // so it replaces rather than accumulates. Without ADD it stays inside the already + // registered set, so measuring never registers a partition the command did not ask for. + 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..6dc61e0f63eb 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,19 @@ case class PaimonRepairFormatTablePartitionsExec( extends LeafV2CommandExec { override protected def run(): Seq[InternalRow] = { + // A repair stops at the numbers a directory listing already gives, so it asks for no record + // count: that one needs the file footers, which is what ANALYZE TABLE pays for. + val statsCollector = + if (OptionUtils.formatTableRepairCollectStatistics()) { + new FormatTablePartitionStatsCollector( + table.table, + false, + 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..6664109e7171 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,140 @@ 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, false, 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, false, 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, false, 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 +516,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 +583,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 +598,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 }