This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 2ce5cea797 [core][spark] Support collecting format table partition
statistics in MSCK REPAIR TABLE (#9297)
2ce5cea797 is described below
commit 2ce5cea797d08785be6d97e0300b3e6950304857
Author: Dapeng Sun(孙大鹏) <[email protected]>
AuthorDate: Fri Aug 21 08:54:45 2026 +0800
[core][spark] Support collecting format table partition statistics in MSCK
REPAIR TABLE (#9297)
---
docs/generated/spark_connector_configuration.html | 12 +
.../format/FormatTablePartitionStatsCollector.java | 195 ++++++++++++
.../FormatTablePartitionStatsCollectorTest.java | 346 +++++++++++++++++++++
.../apache/paimon/spark/SparkConnectorOptions.java | 17 +
.../spark/format/FormatTablePartitionRepair.java | 77 ++++-
.../PaimonFormatTablePartitionDdlExec.scala | 14 +-
.../org/apache/paimon/spark/util/OptionUtils.scala | 8 +
.../format/FormatTablePartitionRepairTest.java | 147 ++++++++-
.../CatalogManagedPartitionMsckRepairTest.scala | 44 ++-
9 files changed, 841 insertions(+), 19 deletions(-)
diff --git a/docs/generated/spark_connector_configuration.html
b/docs/generated/spark_connector_configuration.html
index 875b3d5639..71bb386148 100644
--- a/docs/generated/spark_connector_configuration.html
+++ b/docs/generated/spark_connector_configuration.html
@@ -26,6 +26,18 @@ under the License.
</tr>
</thead>
<tbody>
+ <tr>
+ <td><h5>format-table.repair.collect-statistics</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>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.</td>
+ </tr>
+ <tr>
+ <td><h5>format-table.statistics.parallelism</h5></td>
+ <td style="word-wrap: break-word;">8</td>
+ <td>Integer</td>
+ <td>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.</td>
+ </tr>
<tr>
<td><h5>legacy-timestamp-mapping.enabled</h5></td>
<td style="word-wrap: break-word;">false</td>
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 0000000000..0f2323546b
--- /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.
+ *
+ * <p>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.
+ *
+ * <p>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<PartitionStatistics> collect(List<Map<String, String>>
partitions) {
+ if (partitions.isEmpty()) {
+ return Collections.emptyList();
+ }
+ int threads = Math.min(parallelism, partitions.size());
+ if (threads == 1) {
+ List<PartitionStatistics> statistics = new
ArrayList<>(partitions.size());
+ for (Map<String, String> partition : partitions) {
+ statistics.add(measure(partition));
+ }
+ return statistics;
+ }
+
+ ExecutorService executor =
+ ThreadPoolUtils.createCachedThreadPool(threads,
"FORMAT-TABLE-STATS-THREAD-POOL");
+ try {
+ List<Future<PartitionStatistics>> futures = new
ArrayList<>(partitions.size());
+ for (Map<String, String> partition : partitions) {
+ futures.add(executor.submit(() -> measure(partition)));
+ }
+ List<PartitionStatistics> statistics = new
ArrayList<>(partitions.size());
+ for (Future<PartitionStatistics> 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<String, String> partition) {
+ FileIO fileIO = table.fileIO();
+ Path partitionPath = partitionPath(partition);
+ List<FileStatus> 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<String, String> partition) {
+ return new PartitionStatistics(
+ partition,
+ PartitionStatistics.UNKNOWN,
+ 0L,
+ 0L,
+ PartitionStatistics.UNKNOWN,
+ PartitionStatistics.UNKNOWN_TOTAL_BUCKETS);
+ }
+
+ private Path partitionPath(Map<String, String> partition) {
+ LinkedHashMap<String, String> 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 0000000000..522288c1e0
--- /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.
+ *
+ * <p>The staging cases are the same ones the read path has to survive: a
committer leaves trees
+ * such as {@code _temporary/}, {@code __magic_job-<id>/} 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<String, String> 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<Map<String, String>> partitions =
+ Arrays.asList(spec("2025", "12"), spec("2025", "11"),
spec("2025", "10"));
+
+ List<PartitionStatistics> 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<Map<String, String>> 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<PartitionStatistics> serial =
+ new FormatTablePartitionStatsCollector(table(fileIO,
tablePath), 1)
+ .collect(partitions);
+ List<PartitionStatistics> 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<Map<String, String>> 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<String,
String> 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<String, String> 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<String, String> spec(String year, String month) {
+ LinkedHashMap<String, String> 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 108fe12dac..4c0ec743ff 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<Boolean>
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<Integer>
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<Boolean>
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 d77968f248..d213f4d86b 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.fs.Path;
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 @@ public class 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.
+ *
+ * <p>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 class FormatTablePartitionRepair {
listFilesystemPartitionSpecs(formatTable),
formatTable.partitionKeys(),
addPartitions,
- dropPartitions);
+ dropPartitions,
+ statsCollector);
}
private static List<Map<String, String>>
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<Pair<LinkedHashMap<String, String>, Path>> found =
@@ -96,11 +115,9 @@ public class FormatTablePartitionRepair {
/**
* 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 @@ public class FormatTablePartitionRepair {
List<String> partitionKeys,
boolean addPartitions,
boolean dropPartitions) {
+ return apply(
+ partitionManager,
+ filesystemPartitions,
+ partitionKeys,
+ addPartitions,
+ dropPartitions,
+ null);
+ }
+
+ static int apply(
+ FormatTablePartitionManager partitionManager,
+ List<Map<String, String>> filesystemPartitions,
+ List<String> partitionKeys,
+ boolean addPartitions,
+ boolean dropPartitions,
+ @Nullable FormatTablePartitionStatsCollector statsCollector) {
Set<Map<String, String>> registeredPartitions = new HashSet<>();
for (Partition partition :
partitionManager.listPartitions(Collections.<String,
String>emptyMap(), null)) {
@@ -135,11 +168,25 @@ public class FormatTablePartitionRepair {
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<Map<String, String>> measured = new ArrayList<>();
+ for (Map<String, String> 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 45c925b446..84cd80d12d 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 10d403248b..1649a57ead 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 033285c46e..1ee208f2b9 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.partition.PartitionStatistics;
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 org.junit.jupiter.api.io.TempDir;
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 @@ class FormatTablePartitionRepairTest {
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<PartitionStatistics> 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<String, String> spec(String key, String value) {
Map<String, String> spec = new LinkedHashMap<>();
spec.put(key, value);
@@ -380,13 +513,21 @@ class FormatTablePartitionRepairTest {
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 @@ class FormatTablePartitionRepairTest {
private final List<List<Map<String, String>>> createdPartitions = new
ArrayList<>();
private final List<Boolean> createIgnoreFlags = new ArrayList<>();
private final List<List<Map<String, String>>> droppedPartitions = new
ArrayList<>();
+ private final List<List<PartitionStatistics>> reportedStatistics = new
ArrayList<>();
+ private final List<Boolean> replaceFlags = new ArrayList<>();
private void register(List<Map<String, String>> partitions) {
registered.addAll(partitions);
@@ -452,6 +595,8 @@ class FormatTablePartitionRepairTest {
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 586099b73b..bee92ed562 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
}
