diff --git a/docs/docs/spark/auxiliary.md b/docs/docs/spark/auxiliary.md index c046bf12563c..1ccf9c004b5e 100644 --- a/docs/docs/spark/auxiliary.md +++ b/docs/docs/spark/auxiliary.md @@ -130,6 +130,10 @@ ANALYZE TABLE my_table COMPUTE STATISTICS FOR COLUMNS col1; ANALYZE TABLE my_table COMPUTE STATISTICS FOR ALL COLUMNS; ``` +On a Format Table with catalog-managed partitions the statement means something narrower: it +measures the table's partitions and supports `PARTITION (...)` and `NOSCAN`, see +[Manage Format Table Partitions](./sql-ddl#manage-format-table-partitions). + ## Refresh table The REFRESH TABLE statement invalidates the cached entries, which include data and metadata of the given table. diff --git a/docs/docs/spark/sql-ddl.md b/docs/docs/spark/sql-ddl.md index 4ac28fd6521e..b0901f74ad8f 100644 --- a/docs/docs/spark/sql-ddl.md +++ b/docs/docs/spark/sql-ddl.md @@ -216,15 +216,35 @@ ALTER TABLE my_table ADD PARTITION (dt='2025-01-01'); ALTER TABLE my_table DROP PARTITION (dt='2025-01-01'); MSCK REPAIR TABLE my_table; SHOW PARTITIONS my_table; +ANALYZE TABLE my_table PARTITION (dt='2025-01-01') COMPUTE STATISTICS NOSCAN; ``` On a Format Table whose partitions are discovered from the filesystem, `ADD PARTITION`, -`DROP PARTITION` and `MSCK REPAIR TABLE` fail with an error. +`DROP PARTITION`, `MSCK REPAIR TABLE` and `ANALYZE TABLE` fail with an error. `ADD PARTITION` creates the partition directory and registers the partition; querying a newly added partition before any data is written returns no rows. `DROP PARTITION` unregisters the partition and deletes its directory. +`ANALYZE TABLE` measures partitions. A Format Table has no snapshot to carry a table-level +statistic and no column statistics, so `COMPUTE STATISTICS FOR COLUMNS` and `FOR ALL COLUMNS` are +not supported on it; what the statement writes back to the catalog is the file count, byte size, +last file creation time and row count of the partitions it measured. The measurement replaces +what the catalog held for those partitions, so running it twice reports the same numbers as +running it once, and it never adds or removes a partition — use `MSCK REPAIR TABLE` for that. + +`NOSCAN` stops at the directory listing, which gives everything except the row count. Without it, +the row count is read from each file's footer, so it is exact for the formats that carry one +(Parquet, ORC) and stays unknown for the ones that do not (CSV, TEXT, JSON) rather than being +guessed. Reading footers costs one open per file, so `NOSCAN` is the cheaper of the two. + +A `PARTITION (...)` clause must give values for a leading run of the partition columns, because +that is the shape the catalog can select on. On a table partitioned by `(dt, hh)`, +`PARTITION (dt='2025-01-01')` and `PARTITION (dt='2025-01-01', hh)` both measure every hour of +that day, while `PARTITION (hh='01')` is rejected rather than widened to every day. Naming a +partition that is not registered is an error too, rather than a statement that reports success for +having measured nothing. + :::info `metastore.partitioned-table = true` enables catalog-managed partitions, which requires an diff --git a/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..95f1e2519433 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionStatsCollector.java @@ -0,0 +1,288 @@ +/* + * 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; 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); + + /** Row counts that need no columns: the file footer alone answers how many rows it holds. */ + private static final RowType NO_COLUMNS = RowType.builder().build(); + + private static final SimpleColStatsCollector.Factory[] NO_COLLECTORS = + new SimpleColStatsCollector.Factory[0]; + + private final FormatTable table; + + private final boolean onlyValueInPath; + private final int parallelism; + + private final boolean withRecordCount; + + /** Measures from the listing alone, leaving the record count unknown. */ + public FormatTablePartitionStatsCollector(FormatTable table, int parallelism) { + this(table, false, parallelism); + } + + /** + * Measures from the listing, and when {@code withRecordCount} is set also opens every file's + * footer for the rows it holds. That is the expensive half, so it is asked for rather than + * assumed. + */ + public FormatTablePartitionStatsCollector( + FormatTable table, boolean withRecordCount, int parallelism) { + this.table = table; + this.onlyValueInPath = + new CoreOptions(table.options()).formatTablePartitionOnlyValueInPath(); + this.withRecordCount = withRecordCount; + this.parallelism = Math.max(1, parallelism); + } + + /** + * 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 { + // 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, 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); + } + + /** + * Rows in one file, or unknown when its footer cannot be read. One unreadable file makes the + * whole partition unknown rather than short: a sum missing a file, reported as exact, is worse + * than no number at all. + */ + private long rowCount(SimpleStatsExtractor rowCounter, FileStatus file) { + try { + return rowCounter + .extractWithFileInfo(table.fileIO(), file.getPath(), file.getLen()) + .getRight() + .getRowCount(); + } catch (Exception e) { + LOG.warn( + "Failed to read the row count of {} in table {}; the row count of its " + + "partition stays unknown.", + file.getPath(), + table.fullName(), + e); + return PartitionStatistics.UNKNOWN; + } + } + + /** + * A footer reader for this table's format, or null when the format carries no row count. It is + * built with no columns on purpose: only the file's row count is wanted, and asking for column + * statistics would both cost more and make the reader depend on the file schema matching the + * table's. + */ + @Nullable + private SimpleStatsExtractor rowCounter() { + try { + CoreOptions options = new CoreOptions(table.options()); + Optional extractor = + FileFormat.fileFormat(options).createStatsExtractor(NO_COLUMNS, NO_COLLECTORS); + return extractor.orElse(null); + } catch (Exception e) { + LOG.warn( + "Failed to create a row counter for format {} of table {}; row counts stay " + + "unknown.", + table.format(), + table.fullName(), + e); + return null; + } + } + + /** A partition with nothing in it: the file numbers are an exact zero. */ + private static PartitionStatistics empty( + Map partition, @Nullable SimpleStatsExtractor rowCounter) { + return new PartitionStatistics( + partition, + // Only a measurement that counts rows has learned that this partition holds none; + // a zero from one that never opens a file would outlive the emptiness, since the + // measurement after the files arrive reports unknown and unknown replaces nothing. + rowCounter == null ? PartitionStatistics.UNKNOWN : 0L, + 0L, + 0L, + PartitionStatistics.UNKNOWN, + 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..dd6d3f4a0b25 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTablePartitionStatsCollectorTest.java @@ -0,0 +1,460 @@ +/* + * 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); + + 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 measure(fileIO, tablePath, false); + } + + private PartitionStatistics measure(FileIO fileIO, Path tablePath, boolean withRecordCount) { + return new FormatTablePartitionStatsCollector(table(fileIO, tablePath), withRecordCount, 1) + .collect(Collections.singletonList(spec("2025", "10"))) + .get(0); + } + + private FormatTable table(FileIO fileIO, Path tablePath) { + return table(fileIO, tablePath, FormatTable.Format.CSV, "csv"); + } + + @Test + void testAStagedPlaceholderDoesNotEraseTheRowCount() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + writeParquet(fileIO, tablePath, PARTITION_DIR + "/data-0.parquet", 3); + // The magic committer's zero-byte placeholder, under a data file name. Reading it as a + // data file fails, and a failed read turns the whole partition's row count into an + // unknown, so pruning the staging tree is what keeps the count exact. + write( + fileIO, + tablePath, + PARTITION_DIR + "/__magic_job-1/tasks/attempt_1/__base/part-1.parquet", + 0); + + PartitionStatistics measured = + new FormatTablePartitionStatsCollector(parquetTable(fileIO, tablePath), true, 1) + .collect(Collections.singletonList(spec("2025", "10"))) + .get(0); + + assertThat(measured.recordCount()).isEqualTo(3); + assertThat(measured.fileCount()).isEqualTo(1); + } + + @Test + void testParquetRowCountsAreMeasuredExactly() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + writeParquet(fileIO, tablePath, PARTITION_DIR + "/data-0.parquet", 3); + writeParquet(fileIO, tablePath, PARTITION_DIR + "/data-1.parquet", 5); + + PartitionStatistics measured = + new FormatTablePartitionStatsCollector(parquetTable(fileIO, tablePath), true, 1) + .collect(Collections.singletonList(spec("2025", "10"))) + .get(0); + + assertThat(measured.recordCount()).isEqualTo(8); + assertThat(measured.fileCount()).isEqualTo(2); + assertThat(measured.fileSizeInBytes()).isPositive(); + assertThat(PartitionStatistics.isKnown(measured.lastFileCreationTime())).isTrue(); + } + + @Test + void testAnUnreadableFooterLeavesTheWholePartitionRowCountUnknown() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + writeParquet(fileIO, tablePath, PARTITION_DIR + "/data-0.parquet", 3); + // Real bytes, no readable footer: the shape a truncated upload leaves behind. + write(fileIO, tablePath, PARTITION_DIR + "/data-1.parquet", 16); + + PartitionStatistics measured = + new FormatTablePartitionStatsCollector(parquetTable(fileIO, tablePath), true, 1) + .collect(Collections.singletonList(spec("2025", "10"))) + .get(0); + + // One unreadable footer poisons the whole partition row count: a sum missing a file, + // reported as exact, is worse than no number at all. The other fields stay measured. + assertThat(PartitionStatistics.isKnown(measured.recordCount())).isFalse(); + assertThat(measured.fileCount()).isEqualTo(2); + assertThat(measured.fileSizeInBytes()).isPositive(); + } + + @Test + void testCsvKeepsAnUnknownRowCountEvenWhenAsked() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + write(fileIO, tablePath, PARTITION_DIR + "/data-0.csv", 100); + + PartitionStatistics measured = measure(fileIO, tablePath, true); + + assertThat(measured.fileCount()).isEqualTo(1); + // CSV carries no footer: an unknown row count beats a guessed one. + assertThat(PartitionStatistics.isKnown(measured.recordCount())).isFalse(); + } + + private FormatTable parquetTable(FileIO fileIO, Path tablePath) { + return table(fileIO, tablePath, FormatTable.Format.PARQUET, "parquet"); + } + + private static void writeParquet(FileIO fileIO, Path tablePath, String relativePath, int rows) + throws Exception { + Map options = new HashMap<>(); + options.put(CoreOptions.FILE_FORMAT.key(), "parquet"); + FormatWriterFactory factory = + FileFormat.fileFormat(new CoreOptions(options)) + .createWriterFactory( + RowType.builder().field("id", DataTypes.INT()).build()); + Path path = new Path(tablePath, relativePath); + fileIO.mkdirs(path.getParent()); + if (factory instanceof SupportsDirectWrite) { + FormatWriter writer = ((SupportsDirectWrite) factory).create(fileIO, path, "zstd"); + for (int i = 0; i < rows; i++) { + writer.addElement(GenericRow.of(i)); + } + writer.close(); + } else { + try (PositionOutputStream out = fileIO.newOutputStream(path, false)) { + FormatWriter writer = factory.create(out, "zstd"); + for (int i = 0; i < rows; i++) { + writer.addElement(GenericRow.of(i)); + } + writer.close(); + } + } + } + + private FormatTable table(FileIO fileIO, Path tablePath, Map options) { + return table(fileIO, tablePath, FormatTable.Format.CSV, options); + } + + 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/catalyst/analysis/PaimonAnalysis.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala index 1ecb417abe4d..40276aca1c7f 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala @@ -23,7 +23,8 @@ import org.apache.paimon.spark.SparkTable import org.apache.paimon.spark.catalyst.Compatibility import org.apache.paimon.spark.catalyst.analysis.PaimonRelation.isPaimonTable import org.apache.paimon.spark.catalyst.plans.logical.{PaimonDropPartitions, PaimonHiveDynamicPartitionQuery} -import org.apache.paimon.spark.commands.{PaimonAnalyzeTableColumnCommand, PaimonDynamicPartitionOverwriteCommand, PaimonShowColumnsCommand, SchemaEvolutionHelper} +import org.apache.paimon.spark.commands.{PaimonAnalyzeFormatTablePartitionsCommand, PaimonAnalyzeTableColumnCommand, PaimonDynamicPartitionOverwriteCommand, PaimonShowColumnsCommand, SchemaEvolutionHelper} +import org.apache.paimon.spark.format.PaimonFormatTable import org.apache.paimon.spark.util.OptionUtils import org.apache.paimon.table.FileStoreTable @@ -282,6 +283,15 @@ case class PaimonPostHocResolutionRules(session: SparkSession) extends Rule[Logi } withoutHiveDynamicPartitionMarkers match { + // Spark rejects ANALYZE TABLE for every v2 table, so intercept before it does. Unlike a + // Paimon table, a Format Table has no snapshot to carry statistics and no column statistics + // to compute: analyzing it measures its partitions, for which PARTITION(...) and NOSCAN both + // mean something. Tables using filesystem partition discovery have no catalog to write to + // and fall through to the upstream rejection. + case a @ AnalyzeTable(ResolvedTable(_, _, table: PaimonFormatTable, _), partitionSpec, noScan) + if a.resolved && table.hasCatalogManagedPartitions => + PaimonAnalyzeFormatTablePartitionsCommand(table, partitionSpec, noScan) + case a @ AnalyzeTable( ResolvedTable(catalog, identifier, table: SparkTable, _), partitionSpec, diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeFormatTablePartitionsCommand.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeFormatTablePartitionsCommand.scala new file mode 100644 index 000000000000..18268ce9acc7 --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeFormatTablePartitionsCommand.scala @@ -0,0 +1,119 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.spark.commands + +import org.apache.paimon.spark.format.PaimonFormatTable +import org.apache.paimon.spark.leafnode.PaimonLeafRunnableCommand +import org.apache.paimon.spark.util.OptionUtils +import org.apache.paimon.table.format.FormatTablePartitionStatsCollector + +import org.apache.spark.sql.{Row, SparkSession} + +import java.util.{Map => JMap} + +import scala.collection.JavaConverters._ +import scala.collection.immutable.ListMap + +/** + * Recomputes the catalog statistics of a Format Table with catalog-managed partitions, backing + * `ANALYZE TABLE t [PARTITION(...)] COMPUTE STATISTICS [NOSCAN]`. + * + * The partitions are measured from storage and the result replaces what the catalog holds, so this + * is how a table drifts back into agreement after writers the catalog never saw. `NOSCAN` stops at + * what a directory listing gives — file count, byte size, last file creation time — while a full + * ANALYZE also reads each file footer for its row count. Formats that carry no footer (CSV, TEXT, + * JSON) keep an unknown row count either way rather than a guessed one. + * + * Analyzing is not a way to add or remove partitions: it measures the ones registered at the time + * of the listing and re-registers exactly those. There is no lock between the listing and the + * write, so a partition dropped concurrently can be re-registered with its last measurement — the + * same last-writer-wins window every lock-free partition operation on these tables has. A + * `PARTITION(...)` clause selects a leading subset of them. + */ +case class PaimonAnalyzeFormatTablePartitionsCommand( + v2Table: PaimonFormatTable, + partitionSpec: Map[String, Option[String]], + noScan: Boolean) + extends PaimonLeafRunnableCommand { + + override def run(sparkSession: SparkSession): Seq[Row] = { + val prefix = leadingPrefix(sparkSession) + val partitions = v2Table.partitionManager + .listPartitions(prefix.asJava, null) + .asScala + .map(_.spec().asInstanceOf[JMap[String, String]]) + .toList + + if (partitions.isEmpty && prefix.nonEmpty) { + throw new IllegalArgumentException( + s"Partition ${prefix.map { case (k, v) => s"$k=$v" }.mkString("(", ", ", ")")} of table " + + s"${v2Table.name()} does not exist, so there is nothing to measure.") + } + + if (partitions.nonEmpty) { + val collector = new FormatTablePartitionStatsCollector( + v2Table.table, + !noScan, + OptionUtils.formatTableStatisticsParallelism()) + val statistics = collector.collect(partitions.asJava) + v2Table.partitionManager + .createPartitions(partitions.asJava, true, statistics, true) + } + Seq.empty[Row] + } + + /** + * The values the `PARTITION(...)` clause fixes, as a leading prefix of the partition keys — the + * shape the catalog can select on. + * + * This follows what Spark does with the same clause on a metastore table: column names resolve + * under the session's case sensitivity, a column named without a value means every value of it, + * and the columns that do carry a value have to be a leading run. `PARTITION (dt = 'x', hour)` + * therefore selects every hour of that day and `PARTITION (dt, hour)` selects everything, while + * `PARTITION (hour = '00')` is rejected: the catalog cannot select on a non-leading key, and + * quietly widening it would measure more partitions than were asked for. + */ + private def leadingPrefix(sparkSession: SparkSession): Map[String, String] = { + if (partitionSpec.isEmpty) { + return Map.empty + } + val resolver = sparkSession.sessionState.conf.resolver + val partitionKeys = v2Table.table.partitionKeys().asScala.toSeq + val normalized = partitionSpec.map { + case (key, value) => + val resolved = partitionKeys + .find(partitionKey => resolver(partitionKey, key)) + .getOrElse( + throw new IllegalArgumentException( + s"$key is not a partition column of ${v2Table.name()}, whose partition columns are " + + partitionKeys.mkString("[", ", ", "]"))) + resolved -> value + } + val valueByKey = partitionKeys.map(key => key -> normalized.get(key).flatten) + val prefix = valueByKey.takeWhile(_._2.isDefined) + if (valueByKey.drop(prefix.size).exists(_._2.isDefined)) { + throw new IllegalArgumentException( + s"ANALYZE TABLE ${v2Table.name()} PARTITION must give values for a leading run of its " + + s"partition columns ${partitionKeys.mkString("[", ", ", "]")}, but got " + + partitionSpec.keys.mkString("[", ", ", "]")) + } + // Kept in partition-key order, so a message built from it reads in that order too. + ListMap(prefix.map { case (key, value) => key -> value.get }: _*) + } +} diff --git a/paimon-spark/paimon-spark-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/CatalogManagedPartitionAnalyzeTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionAnalyzeTest.scala new file mode 100644 index 000000000000..1e944e376b20 --- /dev/null +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionAnalyzeTest.scala @@ -0,0 +1,407 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.spark.sql + +import org.apache.paimon.catalog.Identifier +import org.apache.paimon.fs.Path +import org.apache.paimon.partition.{Partition, PartitionStatistics} +import org.apache.paimon.spark.PaimonSparkTestWithRestCatalogBase +import org.apache.paimon.table.FormatTable + +import java.util.Locale + +import scala.collection.JavaConverters._ + +/** + * `ANALYZE TABLE ... PARTITION(...) COMPUTE STATISTICS [NOSCAN]` on a Format Table with + * catalog-managed partitions, held against the semantics Spark and Hive give the same statement on + * a Hive metastore table. + * + * The reference behaviour is what Spark's own `StatisticsSuite` and its `AlterTable*Partition` + * command suites pin: a partition column named without a value means every value of it, a spec + * naming a partition that does not exist is an error rather than a no-op, and partition column + * names resolve the way the rest of Spark resolves identifiers. + */ +class CatalogManagedPartitionAnalyzeTest extends PaimonSparkTestWithRestCatalogBase { + + test("ANALYZE with a full partition spec measures only that partition") { + val tableName = "analyze_full_spec" + withTable(tableName) { + createTable(tableName) + writeCsvPartition(tableName, "20260101", "00", 1) + writeCsvPartition(tableName, "20260101", "01", 2) + repair(tableName) + + sql( + s"ANALYZE TABLE ${qualified(tableName)} PARTITION (dt = '20260101', hour = '00') " + + s"COMPUTE STATISTICS NOSCAN").collect() + + assert(statisticsOf(tableName, "20260101", "00").fileCount() == 1L) + assert(!PartitionStatistics.isKnown(statisticsOf(tableName, "20260101", "01").fileCount())) + } + } + + test("ANALYZE with a leading prefix measures every partition under it") { + val tableName = "analyze_prefix" + withTable(tableName) { + createTable(tableName) + writeCsvPartition(tableName, "20260101", "00", 1) + writeCsvPartition(tableName, "20260101", "01", 2) + writeCsvPartition(tableName, "20260102", "00", 3) + repair(tableName) + + sql( + s"ANALYZE TABLE ${qualified(tableName)} PARTITION (dt = '20260101') " + + s"COMPUTE STATISTICS NOSCAN").collect() + + assert(statisticsOf(tableName, "20260101", "00").fileCount() == 1L) + assert(statisticsOf(tableName, "20260101", "01").fileCount() == 1L) + // The sibling day is outside the prefix and keeps whatever it had. + assert(!PartitionStatistics.isKnown(statisticsOf(tableName, "20260102", "00").fileCount())) + } + } + + test("ANALYZE naming every partition column without a value measures every partition") { + val tableName = "analyze_all_columns" + withTable(tableName) { + createTable(tableName) + writeCsvPartition(tableName, "20260101", "00", 1) + writeCsvPartition(tableName, "20260102", "01", 2) + repair(tableName) + + // Spark and Hive read `PARTITION (dt, hour)` as every value of both columns. + sql(s"ANALYZE TABLE ${qualified(tableName)} PARTITION (dt, hour) COMPUTE STATISTICS NOSCAN") + .collect() + + assert(statisticsOf(tableName, "20260101", "00").fileCount() == 1L) + assert(statisticsOf(tableName, "20260102", "01").fileCount() == 1L) + } + } + + test("ANALYZE naming a trailing column without a value measures the set under the prefix") { + val tableName = "analyze_partial_values" + withTable(tableName) { + createTable(tableName) + writeCsvPartition(tableName, "20260101", "00", 1) + writeCsvPartition(tableName, "20260101", "01", 2) + writeCsvPartition(tableName, "20260102", "00", 3) + repair(tableName) + + // Spark and Hive read `PARTITION (dt = 'x', hour)` as every hour of that day. + sql( + s"ANALYZE TABLE ${qualified(tableName)} PARTITION (dt = '20260101', hour) " + + s"COMPUTE STATISTICS NOSCAN").collect() + + assert(statisticsOf(tableName, "20260101", "00").fileCount() == 1L) + assert(statisticsOf(tableName, "20260101", "01").fileCount() == 1L) + assert(!PartitionStatistics.isKnown(statisticsOf(tableName, "20260102", "00").fileCount())) + } + } + + test("ANALYZE of a partition that does not exist fails instead of measuring nothing") { + val tableName = "analyze_missing_partition" + withTable(tableName) { + createTable(tableName) + writeCsvPartition(tableName, "20260101", "00", 1) + repair(tableName) + + // Succeeding silently tells the caller a partition was measured when none was. + val error = intercept[Exception] { + sql( + s"ANALYZE TABLE ${qualified(tableName)} PARTITION (dt = '20991231') " + + s"COMPUTE STATISTICS NOSCAN").collect() + } + assert(causeMessages(error).contains("20991231"), causeMessages(error)) + assert( + causeMessages(error).contains("does not exist, so there is nothing to measure"), + causeMessages(error)) + } + } + + test("ANALYZE of a non-leading partition column is rejected instead of widened") { + val tableName = "analyze_non_leading" + withTable(tableName) { + createTable(tableName) + writeCsvPartition(tableName, "20260101", "00", 1) + writeCsvPartition(tableName, "20260102", "00", 2) + repair(tableName) + + // The catalog selects on a leading run, so the only way to serve this spec is to measure + // every day that has an hour 00 — more partitions than were asked about. + val error = intercept[Exception] { + sql( + s"ANALYZE TABLE ${qualified(tableName)} PARTITION (hour = '00') " + + s"COMPUTE STATISTICS NOSCAN").collect() + } + assert( + causeMessages(error).contains("leading run of its partition columns [dt, hour]"), + causeMessages(error)) + // Rejected means nothing was measured, in either of the two days the widening would reach. + assert(!PartitionStatistics.isKnown(statisticsOf(tableName, "20260101", "00").fileCount())) + assert(!PartitionStatistics.isKnown(statisticsOf(tableName, "20260102", "00").fileCount())) + } + } + + test("ANALYZE of a column that is not a partition column is rejected") { + val tableName = "analyze_not_a_partition_column" + withTable(tableName) { + createTable(tableName) + writeCsvPartition(tableName, "20260101", "00", 1) + repair(tableName) + + // `payload` is a column of the table but not one the catalog partitions on. + val error = intercept[Exception] { + sql( + s"ANALYZE TABLE ${qualified(tableName)} PARTITION (payload = 'a') " + + s"COMPUTE STATISTICS NOSCAN").collect() + } + assert( + causeMessages(error).contains("payload is not a partition column"), + causeMessages(error)) + // Dropping the name from the spec instead would measure the whole table. + assert(!PartitionStatistics.isKnown(statisticsOf(tableName, "20260101", "00").fileCount())) + } + } + + test("ANALYZE resolves partition column names the way the rest of Spark resolves them") { + val tableName = "analyze_case" + withTable(tableName) { + createTable(tableName) + writeCsvPartition(tableName, "20260101", "00", 1) + repair(tableName) + + sql( + s"ANALYZE TABLE ${qualified(tableName)} PARTITION (DT = '20260101') " + + s"COMPUTE STATISTICS NOSCAN").collect() + + assert(statisticsOf(tableName, "20260101", "00").fileCount() == 1L) + } + } + + test("NOSCAN keeps a row count that is already known") { + val tableName = "analyze_noscan_keeps" + withTable(tableName) { + // A full ANALYZE is the way this suite can put an exact row count in the catalog, so the + // table is parquet; what the NOSCAN below must not do is erase it, however it was learned. + sql(s"""CREATE TABLE $tableName (id INT, payload STRING, dt STRING, hour STRING) + |USING PARQUET + |PARTITIONED BY (dt, hour) + |TBLPROPERTIES ( + | 'format-table.implementation' = 'paimon', + | 'metastore.partitioned-table' = 'true') + |""".stripMargin) + sql(s"INSERT INTO ${qualified(tableName)} VALUES (1, 'a', '20260101', '00')") + sql(s"ANALYZE TABLE ${qualified(tableName)} COMPUTE STATISTICS").collect() + val scanned = statisticsOf(tableName, "20260101", "00") + assert(scanned.recordCount() == 1L, scanned.toString) + + // A listing cannot count rows, but it also learned nothing that contradicts the count that + // is already there. Hive keeps numRows across a NOSCAN for exactly this reason. + sql(s"ANALYZE TABLE ${qualified(tableName)} COMPUTE STATISTICS NOSCAN").collect() + + val afterNoScan = statisticsOf(tableName, "20260101", "00") + assert(afterNoScan.fileCount() == 1L, afterNoScan.toString) + assert(afterNoScan.recordCount() == 1L, afterNoScan.toString) + } + } + + test("a full ANALYZE reads the row count a NOSCAN cannot") { + val tableName = "analyze_footers" + withTable(tableName) { + // Parquet carries a row count in its footer. CSV, which the rest of this suite uses, carries + // none, so it is the format that cannot tell a full ANALYZE apart from a NOSCAN. + sql(s"""CREATE TABLE $tableName (id INT, payload STRING, dt STRING, hour STRING) + |USING PARQUET + |PARTITIONED BY (dt, hour) + |TBLPROPERTIES ( + | 'format-table.implementation' = 'paimon', + | 'metastore.partitioned-table' = 'true') + |""".stripMargin) + sql(s"""INSERT INTO ${qualified(tableName)} + |VALUES (1, 'a', '20260101', '00'), (2, 'b', '20260101', '00'), + | (3, 'c', '20260102', '00') + |""".stripMargin) + // The write registered both partitions and, with reporting off, measured neither. + assert(!PartitionStatistics.isKnown(statisticsOf(tableName, "20260101", "00").recordCount())) + + sql(s"ANALYZE TABLE ${qualified(tableName)} PARTITION (dt = '20260101') COMPUTE STATISTICS") + .collect() + + // No NOSCAN, so the footers were read and the row count is exact. + val scanned = statisticsOf(tableName, "20260101", "00") + assert(scanned.recordCount() == 2L, scanned.toString) + assert(scanned.fileCount() >= 1L, scanned.toString) + + sql( + s"ANALYZE TABLE ${qualified(tableName)} PARTITION (dt = '20260102') " + + s"COMPUTE STATISTICS NOSCAN").collect() + + // NOSCAN measured from the listing alone: the file numbers are there, and the row count is + // still nobody's measurement even though this format could have given one. + val listed = statisticsOf(tableName, "20260102", "00") + assert(listed.fileCount() >= 1L, listed.toString) + assert(listed.fileSizeInBytes() > 0L, listed.toString) + assert(!PartitionStatistics.isKnown(listed.recordCount()), listed.toString) + } + } + + test("ANALYZE run twice reports the same measurement rather than accumulating") { + val tableName = "analyze_idempotent" + withTable(tableName) { + createTable(tableName) + writeCsvPartition(tableName, "20260101", "00", 1) + repair(tableName) + + sql(s"ANALYZE TABLE ${qualified(tableName)} COMPUTE STATISTICS NOSCAN").collect() + val once = statisticsOf(tableName, "20260101", "00") + sql(s"ANALYZE TABLE ${qualified(tableName)} COMPUTE STATISTICS NOSCAN").collect() + val twice = statisticsOf(tableName, "20260101", "00") + + // Anchored, so two runs that both measured nothing cannot pass as two equal measurements. + assert(once.fileCount() == 1L, once.toString) + assert(once.fileSizeInBytes() > 0L, once.toString) + assert(twice.fileCount() == once.fileCount(), s"$once then $twice") + assert(twice.fileSizeInBytes() == once.fileSizeInBytes(), s"$once then $twice") + } + } + + test("ANALYZE does not count files a committer left staged in the partition") { + val tableName = "analyze_staging" + withTable(tableName) { + createTable(tableName) + val partitionPath = writeCsvPartition(tableName, "20260101", "00", 1) + val table = formatTable(tableName) + // What a magic committer leaves behind: a data file name under a staging directory. + val staged = + new Path(new Path(partitionPath, "__magic_job-1/tasks/attempt_1/__base"), "part-9.csv") + table.fileIO().mkdirs(staged.getParent) + table.fileIO().writeFile(staged, "9,staged\n", false) + repair(tableName) + + sql(s"ANALYZE TABLE ${qualified(tableName)} COMPUTE STATISTICS NOSCAN").collect() + + val measured = statisticsOf(tableName, "20260101", "00") + // The reader returns one row from one file; a measurement claiming two is a number no query + // can reproduce. + assert(measured.fileCount() == 1L, measured.toString) + assert(sql(s"SELECT COUNT(*) FROM ${qualified(tableName)}").collect()(0).getLong(0) == 1L) + } + } + + test("ANALYZE measures registered partitions and never changes which exist") { + val tableName = "analyze_partition_set" + withTable(tableName) { + createTable(tableName) + writeCsvPartition(tableName, "20260101", "00", 1) + writeCsvPartition(tableName, "20260102", "00", 2) + repair(tableName) + + sql( + s"ANALYZE TABLE ${qualified(tableName)} PARTITION (dt = '20260101') " + + s"COMPUTE STATISTICS NOSCAN").collect() + + val scoped = statisticsOf(tableName, "20260101", "00") + assert(scoped.fileCount() == 1L, scoped.toString) + assert(scoped.fileSizeInBytes() > 0L, scoped.toString) + // A PARTITION clause scopes the measurement; the sibling keeps whatever it had. + val sibling = statisticsOf(tableName, "20260102", "00") + assert(!PartitionStatistics.isKnown(sibling.fileCount()), sibling.toString) + + sql(s"ANALYZE TABLE ${qualified(tableName)} COMPUTE STATISTICS NOSCAN").collect() + assert(statisticsOf(tableName, "20260102", "00").fileCount() == 1L) + // Analyzing measures partitions, it does not decide which ones exist. + val expected = Set("dt=20260101/hour=00", "dt=20260102/hour=00") + assert(registeredPartitions(tableName) == expected) + assert( + sql(s"SHOW PARTITIONS ${qualified(tableName)}").collect().map(_.getString(0)).toSet == + expected) + } + } + + test("ANALYZE is rejected for a format table discovering partitions from the filesystem") { + val tableName = "analyze_filesystem_partitions" + withTable(tableName) { + sql(s"""CREATE TABLE $tableName (id INT, payload STRING, dt STRING, hour STRING) + |USING CSV + |PARTITIONED BY (dt, hour) + |TBLPROPERTIES ( + | 'format-table.implementation' = 'paimon', + | 'metastore.partitioned-table' = 'false') + |""".stripMargin) + + // There is no catalog to write a measurement to, so the table is not intercepted at all and + // keeps Spark's own rejection rather than quietly measuring nothing. + val error = intercept[Exception] { + sql(s"ANALYZE TABLE ${qualified(tableName)} COMPUTE STATISTICS NOSCAN").collect() + } + val messages = causeMessages(error) + assert(messages.contains("ANALYZE TABLE"), messages) + assert(messages.toLowerCase(Locale.ROOT).contains("not supported"), messages) + } + } + + private def qualified(tableName: String): String = s"paimon.$dbName0.$tableName" + + private def createTable(tableName: String): Unit = { + sql(s"""CREATE TABLE $tableName (id INT, payload STRING, dt STRING, hour STRING) + |USING CSV + |PARTITIONED BY (dt, hour) + |TBLPROPERTIES ( + | 'format-table.implementation' = 'paimon', + | 'metastore.partitioned-table' = 'true') + |""".stripMargin) + } + + private def repair(tableName: String): Unit = + sql(s"MSCK REPAIR TABLE ${qualified(tableName)}").collect() + + private def formatTable(tableName: String): FormatTable = + paimonCatalog.getTable(Identifier.create(dbName0, tableName)).asInstanceOf[FormatTable] + + private def writeCsvPartition(tableName: String, dt: String, hour: String, id: Int): Path = { + val table = formatTable(tableName) + val partitionPath = new Path(table.location(), s"dt=$dt/hour=$hour") + table.fileIO().mkdirs(partitionPath) + table + .fileIO() + .writeFile(new Path(partitionPath, f"part-$id%05d.csv"), s"$id,payload-$id\n", false) + partitionPath + } + + private def registeredPartitions(tableName: String): Set[String] = + paimonCatalog + .listPartitions(Identifier.create(dbName0, tableName)) + .asScala + .map(partition => s"dt=${partition.spec().get("dt")}/hour=${partition.spec().get("hour")}") + .toSet + + private def statisticsOf(tableName: String, dt: String, hour: String): Partition = + paimonCatalog + .listPartitions(Identifier.create(dbName0, tableName)) + .asScala + .find(p => p.spec().get("dt") == dt && p.spec().get("hour") == hour) + .getOrElse(fail(s"partition dt=$dt/hour=$hour of $tableName is not registered")) + + private def causeMessages(error: Throwable): String = + Iterator + .iterate(error)(_.getCause) + .takeWhile(_ != null) + .map(e => String.valueOf(e.getMessage)) + .mkString(" | ") +} 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 }