From 206bb1efd9f644970bca40dcde31f6f645d1b457 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dapeng=20Sun=28=E5=AD=99=E5=A4=A7=E9=B9=8F=29?= Date: Tue, 25 Aug 2026 00:42:00 +0800 Subject: [PATCH 1/2] [api] Decode absent partition statistics as unknown instead of zero A listPartitions response may omit the statistics, since the REST contract does not mark them required. The primitive JsonCreator parameters decoded that absence to 0, which PartitionStatistics.isKnown reads as an exact measurement. Since #9351 the zero reaches Spark as a 0 row and 0 byte scan, and a large format table can then be picked as a broadcast build side. --- .../apache/paimon/partition/Partition.java | 63 +++++++++++++++---- .../paimon/partition/PartitionTest.java | 59 +++++++++++++++++ .../CatalogManagedPartitionScanTest.java | 26 ++++++++ 3 files changed, 135 insertions(+), 13 deletions(-) diff --git a/paimon-api/src/main/java/org/apache/paimon/partition/Partition.java b/paimon-api/src/main/java/org/apache/paimon/partition/Partition.java index c9fa4e099752..ad8d0d5cc73c 100644 --- a/paimon-api/src/main/java/org/apache/paimon/partition/Partition.java +++ b/paimon-api/src/main/java/org/apache/paimon/partition/Partition.java @@ -74,20 +74,19 @@ public class Partition extends PartitionStatistics { @Nullable private final Map options; - @JsonCreator public Partition( - @JsonProperty(FIELD_SPEC) Map spec, - @JsonProperty(FIELD_RECORD_COUNT) long recordCount, - @JsonProperty(FIELD_FILE_SIZE_IN_BYTES) long fileSizeInBytes, - @JsonProperty(FIELD_FILE_COUNT) long fileCount, - @JsonProperty(FIELD_LAST_FILE_CREATION_TIME) long lastFileCreationTime, - @JsonProperty(FIELD_TOTAL_BUCKETS) int totalBuckets, - @JsonProperty(FIELD_DONE) boolean done, - @JsonProperty(FIELD_CREATED_AT) @Nullable Long createdAt, - @JsonProperty(FIELD_CREATED_BY) @Nullable String createdBy, - @JsonProperty(FIELD_UPDATED_AT) @Nullable Long updatedAt, - @JsonProperty(FIELD_UPDATED_BY) @Nullable String updatedBy, - @JsonProperty(FIELD_OPTIONS) @Nullable Map options) { + Map spec, + long recordCount, + long fileSizeInBytes, + long fileCount, + long lastFileCreationTime, + int totalBuckets, + boolean done, + @Nullable Long createdAt, + @Nullable String createdBy, + @Nullable Long updatedAt, + @Nullable String updatedBy, + @Nullable Map options) { super(spec, recordCount, fileSizeInBytes, fileCount, lastFileCreationTime, totalBuckets); this.done = done; this.createdAt = createdAt; @@ -120,6 +119,44 @@ public Partition( null); } + /** + * Reads a partition off the wire. The statistics are optional in the REST contract, so an + * absent one decodes to {@link #UNKNOWN} rather than to {@code 0}. {@code totalBuckets} keeps + * its {@code 0} default, for writers older than that field. + */ + @JsonCreator + static Partition fromJson( + @JsonProperty(FIELD_SPEC) Map spec, + @JsonProperty(FIELD_RECORD_COUNT) @Nullable Long recordCount, + @JsonProperty(FIELD_FILE_SIZE_IN_BYTES) @Nullable Long fileSizeInBytes, + @JsonProperty(FIELD_FILE_COUNT) @Nullable Long fileCount, + @JsonProperty(FIELD_LAST_FILE_CREATION_TIME) @Nullable Long lastFileCreationTime, + @JsonProperty(FIELD_TOTAL_BUCKETS) int totalBuckets, + @JsonProperty(FIELD_DONE) boolean done, + @JsonProperty(FIELD_CREATED_AT) @Nullable Long createdAt, + @JsonProperty(FIELD_CREATED_BY) @Nullable String createdBy, + @JsonProperty(FIELD_UPDATED_AT) @Nullable Long updatedAt, + @JsonProperty(FIELD_UPDATED_BY) @Nullable String updatedBy, + @JsonProperty(FIELD_OPTIONS) @Nullable Map options) { + return new Partition( + spec, + orUnknown(recordCount), + orUnknown(fileSizeInBytes), + orUnknown(fileCount), + orUnknown(lastFileCreationTime), + totalBuckets, + done, + createdAt, + createdBy, + updatedAt, + updatedBy, + options); + } + + private static long orUnknown(@Nullable Long value) { + return value == null ? UNKNOWN : value; + } + @JsonGetter(FIELD_DONE) public boolean done() { return done; diff --git a/paimon-api/src/test/java/org/apache/paimon/partition/PartitionTest.java b/paimon-api/src/test/java/org/apache/paimon/partition/PartitionTest.java index b589a75cf1b6..ce514caae991 100644 --- a/paimon-api/src/test/java/org/apache/paimon/partition/PartitionTest.java +++ b/paimon-api/src/test/java/org/apache/paimon/partition/PartitionTest.java @@ -61,6 +61,65 @@ void testJsonSerializationWithNullValues() { assertThat(json).contains("totalBuckets"); } + @Test + void testAbsentStatisticsAreUnknownNotZero() { + // What listPartitions returns from a catalog that stores no statistics. Every consumer of + // this response reads the numbers through PartitionStatistics.isKnown, so absence has to + // arrive as unknown and not as an exact zero. + String statisticsFreeJson = "{\"spec\":{\"pt\":\"1\"},\"done\":true}"; + + Partition partition = JsonSerdeUtil.fromJson(statisticsFreeJson, Partition.class); + + assertThat(partition.spec()).containsEntry("pt", "1"); + assertThat(partition.done()).isTrue(); + assertThat(partition.recordCount()).isEqualTo(PartitionStatistics.UNKNOWN); + assertThat(partition.fileSizeInBytes()).isEqualTo(PartitionStatistics.UNKNOWN); + assertThat(partition.fileCount()).isEqualTo(PartitionStatistics.UNKNOWN); + assertThat(partition.lastFileCreationTime()).isEqualTo(PartitionStatistics.UNKNOWN); + assertThat(partition.createdAt()).isNull(); + assertThat(partition.options()).isNull(); + } + + @Test + void testReportedZeroStaysAnExactMeasurement() { + // The other half of the same boundary: a partition someone measured as empty must not come + // back as unknown. + String emptyPartitionJson = + "{\"spec\":{\"pt\":\"1\"},\"recordCount\":0,\"fileSizeInBytes\":0," + + "\"fileCount\":0,\"lastFileCreationTime\":0}"; + + Partition partition = JsonSerdeUtil.fromJson(emptyPartitionJson, Partition.class); + + assertThat(partition.recordCount()).isEqualTo(0L); + assertThat(partition.fileSizeInBytes()).isEqualTo(0L); + assertThat(partition.fileCount()).isEqualTo(0L); + assertThat(PartitionStatistics.isKnown(partition.recordCount())).isTrue(); + } + + @Test + void testMeasurementsSurviveARoundTrip() { + Partition partition = + new Partition( + Collections.singletonMap("pt", "1"), + 0L, // an empty partition someone did measure + 1024L, + 2L, + 1234567890L, + 10, + true, + 1234567890L, + "user1", + 1234567900L, + "user2", + Collections.singletonMap("key", "value")); + + Partition parsed = + JsonSerdeUtil.fromJson(JsonSerdeUtil.toFlatJson(partition), Partition.class); + + assertThat(parsed).isEqualTo(partition); + assertThat(parsed.recordCount()).isEqualTo(0L); + } + @Test void testJsonSerializationWithNonNullValues() { Map spec = Collections.singletonMap("pt", "1"); diff --git a/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java b/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java index 5ce899694db6..2563a54b354a 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java @@ -37,6 +37,7 @@ import org.apache.paimon.table.source.Split; import org.apache.paimon.types.DataTypes; import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.JsonSerdeUtil; import org.apache.paimon.utils.PartitionPathUtils; import org.junit.jupiter.api.DisplayName; @@ -206,6 +207,31 @@ void testPlanReusesCatalogListingForSplitsAndRowCount() throws Exception { verify(catalog).listPartitionsPaged(IDENTIFIER, 1000, null, null); } + @Test + void testPlanRowCountStaysUnknownWhenCatalogReportsNoStatistics() throws Exception { + // Partitions as they come off the wire from a catalog that stores no statistics. Summing + // them as zeros would tell Spark the table is empty and get a huge scan broadcast. + Catalog catalog = mock(Catalog.class); + Partition october = + JsonSerdeUtil.fromJson( + "{\"spec\":{\"year\":\"2025\",\"month\":\"10\"}}", Partition.class); + Partition november = + JsonSerdeUtil.fromJson( + "{\"spec\":{\"year\":\"2025\",\"month\":\"11\"}}", Partition.class); + when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull())) + .thenReturn(new PagedList<>(Arrays.asList(october, november), null)); + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + writeDataFile(fileIO, tablePath, "year=2025/month=10"); + writeDataFile(fileIO, tablePath, "year=2025/month=11"); + FormatTable table = createTable(fileIO, tablePath, partitionManager(catalog), false); + + FormatTableScan.Plan plan = new FormatTableScan(table, null, null).plan(); + + assertThat(plan.splits()).hasSize(2); + assertThat(plan.rowCount()).isEqualTo(OptionalLong.empty()); + } + @Test void testUnderscoreInPartitionNameRemainsLiteralPrefix() { LocalFileIO fileIO = LocalFileIO.create(); From cac4e9921ec6b52d62fc32003b6b002e0b9d4f79 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dapeng=20Sun=28=E5=AD=99=E5=A4=A7=E9=B9=8F=29?= Date: Tue, 25 Aug 2026 00:42:02 +0800 Subject: [PATCH 2/2] [spark] Weigh files when a partition reports zero rows over real bytes A catalog that cannot tell "never measured" from "measured, and empty" reports a well formed zero. Since #9351 that becomes the scan row count, and sizeInBytes follows it down because the row estimate branch is taken whenever numRows is present, leaving the file size fallback unreachable. The scan then enters the optimizer at 0 bytes and can be picked as a broadcast build side. --- .../paimon/spark/read/PaimonStatistics.scala | 25 +++++++----- .../CatalogManagedPartitionAnalyzeTest.scala | 39 +++++++++++++++++++ 2 files changed, 55 insertions(+), 9 deletions(-) diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/PaimonStatistics.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/PaimonStatistics.scala index 9b06223aeece..9df3e683b4bc 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/PaimonStatistics.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/PaimonStatistics.scala @@ -43,9 +43,17 @@ case class PaimonStatistics( scanRowCount: OptionalLong = OptionalLong.empty() ) extends Statistics { + private lazy val fileTotalSize: Long = splits.map(SplitUtils.splitSize).sum + lazy val numRows: OptionalLong = { if (scanRowCount.isPresent) { - scanRowCount + // A catalog may report zero because it cannot tell "never measured" from "measured, and + // empty". Over real bytes leave it unknown; over zero bytes the scan really is empty. + if (scanRowCount.getAsLong > 0 || fileTotalSize == 0) { + scanRowCount + } else { + OptionalLong.empty() + } } else if (splits.exists(_.rowCount() == -1)) { OptionalLong.empty() } else { @@ -54,20 +62,19 @@ case class PaimonStatistics( } lazy val sizeInBytes: OptionalLong = { - if (numRows.isPresent) { + if (numRows.isPresent && numRows.getAsLong > 0) { val sizeInBytes = numRows.getAsLong * estimateRowSize(readRowType) // Avoid return 0 bytes if there are some valid rows. // Avoid return too small size in bytes which may less than row count, // note the compression ratio on disk is usually bigger than memory. OptionalLong.of(Math.max(sizeInBytes, numRows.getAsLong)) + } else if (fileTotalSize > 0) { + // Zero rows times any row size is zero, so weigh the files instead. + OptionalLong.of((fileTotalSize * readRowSizeRatio).toLong) + } else if (numRows.isPresent) { + OptionalLong.of(0L) } else { - val fileTotalSize = splits.map(SplitUtils.splitSize).sum - if (fileTotalSize == 0) { - OptionalLong.empty() - } else { - val size = (fileTotalSize * readRowSizeRatio).toLong - OptionalLong.of(size) - } + OptionalLong.empty() } } 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 index cd3193ab4226..e2e85ae64e7b 100644 --- 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 @@ -376,6 +376,45 @@ class CatalogManagedPartitionAnalyzeTest extends PaimonSparkTestWithRestCatalogB } } + test("a zero row count does not make a partition that still has files look free to read") { + val tableName = "analyze_zero_row_count_size" + withTable(tableName) { + createTable(tableName) + sql( + s"INSERT INTO ${qualified(tableName)} VALUES " + + s"(1, 'a', '20260101', '00'), (2, 'b', '20260101', '00')") + + // Positive control: a real measurement gives both a row count and a size. + val measured = getFormatTableScan(s"SELECT * FROM ${qualified(tableName)}").estimateStatistics + assert(measured.numRows().getAsLong == 2L) + assert(measured.sizeInBytes().getAsLong > 0L) + + // Rewrite the statistics to zero without touching the files, the way a catalog that cannot + // tell "never measured" from "measured, and empty" answers. + val spec = Map("dt" -> "20260101", "hour" -> "00").asJava + paimonCatalog.createPartitions( + Identifier.create(dbName0, tableName), + List(spec).asJava, + true, + List( + new PartitionStatistics( + spec, + 0L, + 0L, + 0L, + 0L, + PartitionStatistics.UNKNOWN_TOTAL_BUCKETS)).asJava, + true + ) + + val zeroed = getFormatTableScan(s"SELECT * FROM ${qualified(tableName)}").estimateStatistics + // Neither number may say the scan is free: the size drives broadcast, the row count drives + // join reordering. + assert(!zeroed.numRows().isPresent, zeroed.numRows().toString) + assert(zeroed.sizeInBytes().getAsLong > 0L, zeroed.sizeInBytes().toString) + } + } + test("partition row count is not duplicated across format data splits") { val tableName = "analyze_multi_split_statistics" withTable(tableName) {