This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 53f0a24d5f [api][spark] Treat an unreported partition statistic as
unknown, not as zero (#9379)
53f0a24d5f is described below
commit 53f0a24d5f180f1516fd0c34ddec79d1c5dc9045
Author: Dapeng Sun(孙大鹏) <[email protected]>
AuthorDate: Tue Aug 25 08:49:45 2026 +0800
[api][spark] Treat an unreported partition statistic as unknown, not as
zero (#9379)
---
.../org/apache/paimon/partition/Partition.java | 63 +++++++++++++++++-----
.../org/apache/paimon/partition/PartitionTest.java | 59 ++++++++++++++++++++
.../format/CatalogManagedPartitionScanTest.java | 26 +++++++++
.../paimon/spark/read/PaimonStatistics.scala | 25 +++++----
.../sql/CatalogManagedPartitionAnalyzeTest.scala | 39 ++++++++++++++
5 files changed, 190 insertions(+), 22 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 c9fa4e0997..ad8d0d5cc7 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<String, String> options;
- @JsonCreator
public Partition(
- @JsonProperty(FIELD_SPEC) Map<String, String> 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<String, String>
options) {
+ Map<String, String> 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<String, String> options) {
super(spec, recordCount, fileSizeInBytes, fileCount,
lastFileCreationTime, totalBuckets);
this.done = done;
this.createdAt = createdAt;
@@ -120,6 +119,44 @@ public class Partition extends PartitionStatistics {
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<String, String> 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<String, String>
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 b589a75cf1..ce514caae9 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 @@ class PartitionTest {
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<String, String> 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 5ce899694d..2563a54b35 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.FormatTable;
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 @@ class CatalogManagedPartitionScanTest {
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();
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 9b06223aee..9df3e683b4 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 cd3193ab42..e2e85ae64e 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) {
