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 }